FS-4996 --resolve

diff --git a/src/mod/event_handlers/mod_erlang_event/handle_msg.c b/src/mod/event_handlers/mod_erlang_event/handle_msg.c
index 618b7eb829e60d2e6af366f59ce48334d6412fe5..21e7376e98cc2c27568e5c09286210e4a50111ae 100644 (file)
--- a/src/mod/event_handlers/mod_erlang_event/handle_msg.c
+++ b/src/mod/event_handlers/mod_erlang_event/handle_msg.c
@@ -660,11 +660,13 @@ static switch_status_t handle_msg_api(listener_t *listener, erlang_msg * msg, in
        }
 }
 
+#define ARGLEN 2048
 static switch_status_t handle_msg_bgapi(listener_t *listener, erlang_msg * msg, int arity, ei_x_buff * buf, ei_x_buff * rbuf)
 {
        char api_cmd[MAXATOMLEN];
-       char arg[1024];
-       if (arity < 3 || ei_decode_atom(buf->buff, &buf->index, api_cmd) || ei_decode_string_or_binary(buf->buff, &buf->index, 1023, arg)) {
+       char arg[ARGLEN];
+
+       if (arity < 3 || ei_decode_atom(buf->buff, &buf->index, api_cmd) || ei_decode_string_or_binary(buf->buff, &buf->index, ARGLEN - 1, arg)) {
                ei_x_encode_tuple_header(rbuf, 2);
                ei_x_encode_atom(rbuf, "error");
                ei_x_encode_atom(rbuf, "badarg");
@@ -1146,6 +1148,7 @@ static switch_status_t handle_ref_tuple(listener_t *listener, erlang_msg * msg,
                        se->spawn_reply->pid = switch_core_alloc(se->pool, sizeof(erlang_pid));
                        switch_assert(se->spawn_reply->pid != NULL);
                        memcpy(se->spawn_reply->pid, &pid, sizeof(erlang_pid));
+
                        switch_thread_cond_signal(se->spawn_reply->ready_or_found);
 
                        switch_mutex_unlock(se->spawn_reply->mutex);

diff --git a/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c b/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c
index 9b34bfa8fc5dfe0c9b58c93fcf4b8b159a2f45d5..bf5b0b6f7166eace7c958a4b566e8cb30002a623 100644 (file)
--- a/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c
+++ b/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c
@@ -330,11 +330,12 @@ static void destroy_session_elem(session_elem_t *session_element)
        switch_thread_rwlock_wrlock(session_element->rwlock);
        switch_thread_rwlock_unlock(session_element->rwlock);
 
-       if ((session = switch_core_session_locate(session_element->uuid_str))) {
+       if ((session = switch_core_session_force_locate(session_element->uuid_str))) {
                switch_channel_t *channel = switch_core_session_get_channel(session);
 
                switch_channel_set_private(channel, "_erlang_session_", NULL);
                switch_channel_clear_flag(channel, CF_CONTROLLED);
+               switch_core_session_soft_unlock(session);
                switch_core_session_rwunlock(session);
        }
        switch_core_destroy_memory_pool(&session_element->pool);
@@ -795,21 +796,8 @@ static void handle_exit(listener_t *listener, erlang_pid * pid)
        remove_binding(NULL, pid);      /* TODO - why don't we pass the listener as the first argument? */
 
        if ((s = find_session_elem_by_pid(listener, pid))) {
-               switch_core_session_t *session = NULL;
-
-               if ((session = switch_core_session_locate(s->uuid_str))) {
-                       switch_channel_t *channel = switch_core_session_get_channel(session);
-
-                       if (switch_channel_get_state(channel) < CS_HANGUP) {
-                               switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Outbound session exited unexpectedly %s!\n", s->uuid_str);
-                       }
-
-                       switch_channel_hangup(channel, SWITCH_CAUSE_DESTINATION_OUT_OF_ORDER);
-                       switch_core_session_rwunlock(session);
-               }
-
+               switch_set_flag_locked(s, LFLAG_SESSION_COMPLETE);
                switch_thread_rwlock_unlock(s->rwlock);
-
        }
 
        if (listener->log_process.type == ERLANG_PID && !ei_compare_pids(&listener->log_process.pid, pid)) {
@@ -1295,9 +1283,6 @@ static switch_status_t state_handler(switch_core_session_t *session)
                         * we can throw this away */
                        switch_set_flag_locked(session_element, LFLAG_SESSION_COMPLETE);
                }
-       } else {
-               switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "unable to update channel state for %s to %s\n", switch_core_session_get_uuid(session),
-                                                 switch_channel_state_name(state));
        }
 
        return SWITCH_STATUS_SUCCESS;
@@ -1307,7 +1292,7 @@ session_elem_t *session_elem_create(listener_t *listener, switch_core_session_t
 {
        /* create a session list element */
        switch_memory_pool_t *session_elem_pool;
-       session_elem_t *session_element;        /* = malloc(sizeof(*session_element)); */
+       session_elem_t *session_element;
        switch_channel_t *channel = switch_core_session_get_channel(session);
        int x;
 
@@ -1339,6 +1324,7 @@ session_elem_t *session_elem_create(listener_t *listener, switch_core_session_t
        session_element->event_list[SWITCH_EVENT_ALL] = 1; /* defaults to everything */
 
        switch_channel_set_private(channel, "_erlang_session_", session_element);
+       switch_core_session_soft_lock(session, 5);
 
        switch_core_event_hook_add_state_change(session, state_handler);
 
@@ -1399,7 +1385,8 @@ session_elem_t *attach_call_to_spawned_process(listener_t *listener, char *modul
        /* attach the session to the listener */
        add_session_elem_to_listener(listener, session_element);
 
-
+       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Added session to listener\n");
+       
        if (!strcmp(function, "!")) {
                /* send a message to request a pid */
                ei_x_buff rbuf;
@@ -1421,14 +1408,20 @@ session_elem_t *attach_call_to_spawned_process(listener_t *listener, char *modul
        } else {
                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "rpc call: %s:%s(Ref)\n", module, function);
                /* should lock with mutex? */
+               switch_mutex_lock(listener->sock_mutex);
                ei_pid_from_rpc(listener->ec, listener->sockfd, &ref, module, function);
+               switch_mutex_unlock(listener->sock_mutex);
                /*
                   char *argv[1];
                   ei_spawn(listener->ec, listener->sockfd, &ref, module, function, 0, argv);
                 */
        }
 
+
+       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Waiting for reply %s %s\n", hash, session_element->uuid_str);
+       switch_mutex_lock(p->mutex);
        switch_thread_cond_timedwait(p->ready_or_found, p->mutex, 5000000);
+       switch_mutex_unlock(p->mutex);
        if (!p->pid) {
                switch_channel_t *channel = switch_core_session_get_channel(session);
 
@@ -1534,16 +1527,23 @@ SWITCH_STANDARD_APP(erlang_outbound_function)
                }
        } else {
                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Using existing listener for session\n");
+
        }
 
        if (listener) {
 
-               if (module && function) {
-                       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new spawned session for listener\n");
-                       session_element = attach_call_to_spawned_process(listener, module, function, session);
+               if ((session_element = find_session_elem_by_uuid(listener, switch_core_session_get_uuid(session)))) {
+                       switch_thread_rwlock_unlock(session_element->rwlock);
+                       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Session already exists\n");
+
                } else {
-                       switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new registered session for listener\n");
-                       session_element = attach_call_to_registered_process(listener, reg_name, session);
+                       if (module && function) {
+                               switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new spawned session for listener\n");
+                               session_element = attach_call_to_spawned_process(listener, module, function, session);
+                       } else {
+                               switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "Creating new registered session for listener\n");
+                               session_element = attach_call_to_registered_process(listener, reg_name, session);
+                       }
                }
 
                switch_thread_rwlock_unlock(listener->rwlock);
@@ -1552,8 +1552,6 @@ SWITCH_STANDARD_APP(erlang_outbound_function)
                        switch_ivr_park(session, NULL);
                }
 
-       } else {
-               switch_thread_rwlock_unlock(globals.listener_rwlock);
        }
 
        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "exit erlang_outbound_function\n");