push events into a separate thread for processing

git-svn-id: http://svn.freeswitch.org/svn/freeswitch/trunk@7888 d0543943-73ff-0310-b7d9-9358b9ac24b2

diff --git a/src/mod/endpoints/mod_sofia/mod_sofia.c b/src/mod/endpoints/mod_sofia/mod_sofia.c
index a4dab15f8e730a1bbf4849e6079a4bfaf4f997f6..dceef86437814995310e6f2fe9e397abe327d1b4 100644 (file)
--- a/src/mod/endpoints/mod_sofia/mod_sofia.c
+++ b/src/mod/endpoints/mod_sofia/mod_sofia.c
@@ -48,8 +48,6 @@ static char silence_data[13] = "";
 
 #define STRLEN 15
 
-static switch_memory_pool_t *module_pool = NULL;
-
 static switch_status_t sofia_on_init(switch_core_session_t *session);
 
 static switch_status_t sofia_on_loopback(switch_core_session_t *session);
@@ -1874,21 +1872,24 @@ SWITCH_MODULE_LOAD_FUNCTION(mod_sofia_load)
        silence_frame.buflen = sizeof(silence_data);
        silence_frame.flags = SFF_CNG;
 
-       module_pool = pool;
-
        memset(&mod_sofia_globals, 0, sizeof(mod_sofia_globals));
-       switch_mutex_init(&mod_sofia_globals.mutex, SWITCH_MUTEX_NESTED, module_pool);
+       mod_sofia_globals.pool = pool;
+       switch_mutex_init(&mod_sofia_globals.mutex, SWITCH_MUTEX_NESTED, mod_sofia_globals.pool);
 
        switch_find_local_ip(mod_sofia_globals.guess_ip, sizeof(mod_sofia_globals.guess_ip), AF_INET);
 
-       switch_core_hash_init(&mod_sofia_globals.profile_hash, module_pool);
-       switch_core_hash_init(&mod_sofia_globals.gateway_hash, module_pool);
-       switch_mutex_init(&mod_sofia_globals.hash_mutex, SWITCH_MUTEX_NESTED, module_pool);
+       switch_core_hash_init(&mod_sofia_globals.profile_hash, mod_sofia_globals.pool);
+       switch_core_hash_init(&mod_sofia_globals.gateway_hash, mod_sofia_globals.pool);
+       switch_mutex_init(&mod_sofia_globals.hash_mutex, SWITCH_MUTEX_NESTED, mod_sofia_globals.pool);
        
        switch_mutex_lock(mod_sofia_globals.mutex);
        mod_sofia_globals.running = 1;
        switch_mutex_unlock(mod_sofia_globals.mutex);
 
+       switch_queue_create(&mod_sofia_globals.presence_queue, 500000, mod_sofia_globals.pool);
+       switch_queue_create(&mod_sofia_globals.mwi_queue, 500000, mod_sofia_globals.pool);
+       sofia_presence_event_thread_start();
+
        if (config_sofia(0, NULL) != SWITCH_STATUS_SUCCESS) {
                mod_sofia_globals.running = 0;
                return SWITCH_STATUS_GENERR;
@@ -1970,8 +1971,10 @@ SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_sofia_shutdown)
 
        su_deinit();
 
+       switch_mutex_lock(mod_sofia_globals.hash_mutex);
        switch_core_hash_destroy(&mod_sofia_globals.profile_hash);
        switch_core_hash_destroy(&mod_sofia_globals.gateway_hash);
+       switch_mutex_unlock(mod_sofia_globals.hash_mutex);
 
        return SWITCH_STATUS_SUCCESS;
 }

diff --git a/src/mod/endpoints/mod_sofia/mod_sofia.h b/src/mod/endpoints/mod_sofia/mod_sofia.h
index 4e8bdeb32e8d1b0fbdf4def454c2c32b31e51d5e..59852ddbc3ac5fb28e01371891fc9c6403219947 100644 (file)
--- a/src/mod/endpoints/mod_sofia/mod_sofia.h
+++ b/src/mod/endpoints/mod_sofia/mod_sofia.h
@@ -177,6 +177,7 @@ typedef enum {
 } TFLAGS;
 
 struct mod_sofia_globals {
+       switch_memory_pool_t *pool;
        switch_hash_t *profile_hash;
        switch_hash_t *gateway_hash;
        switch_mutex_t *hash_mutex;
@@ -185,6 +186,8 @@ struct mod_sofia_globals {
        int32_t threads;
        switch_mutex_t *mutex;
        char guess_ip[80];
+       switch_queue_t *presence_queue;
+       switch_queue_t *mwi_queue;
 };
 extern struct mod_sofia_globals mod_sofia_globals;
 
@@ -598,3 +601,4 @@ const char *sofia_glue_get_unknown_header(sip_t const *sip, const char *name);
 switch_status_t sofia_glue_build_crypto(private_object_t *tech_pvt, int index, switch_rtp_crypto_key_type_t type, switch_rtp_crypto_direction_t direction);
 void sofia_glue_tech_patch_sdp(private_object_t *tech_pvt);
 switch_status_t sofia_glue_tech_proxy_remote_addr(private_object_t *tech_pvt);
+void sofia_presence_event_thread_start(void);

diff --git a/src/mod/endpoints/mod_sofia/sofia.c b/src/mod/endpoints/mod_sofia/sofia.c
index c59708d54d0e7842d92ce1db6cef2229094fd75d..7936cb588f6f7d2a6804d5b7321b9cb80551a647 100644 (file)
--- a/src/mod/endpoints/mod_sofia/sofia.c
+++ b/src/mod/endpoints/mod_sofia/sofia.c
@@ -2499,7 +2499,7 @@ void sofia_handle_sip_i_invite(nua_t *nua, sofia_profile_t *profile, nua_handle_
 
        
        if (!sofia_endpoint_interface || !(session = switch_core_session_request(sofia_endpoint_interface, NULL))) {
-               nua_respond(nh, SIP_503_SERVICE_UNAVAILABLE, TAG_END());
+               nua_respond(nh, 480, "Maximum Calls In Progress", SIPTAG_RETRY_AFTER_STR("300"), TAG_END());
                return;
        }
 

diff --git a/src/mod/endpoints/mod_sofia/sofia_presence.c b/src/mod/endpoints/mod_sofia/sofia_presence.c
index f58661cbda098267bf2a61aebe999fb4f3ffdad9..408834ec8005a816b530975635eb258197c48b0b 100644 (file)
--- a/src/mod/endpoints/mod_sofia/sofia_presence.c
+++ b/src/mod/endpoints/mod_sofia/sofia_presence.c
@@ -40,6 +40,8 @@ static int sofia_presence_sub_reg_callback(void *pArg, int argc, char **argv, ch
 static int sofia_presence_resub_callback(void *pArg, int argc, char **argv, char **columnNames);
 static int sofia_presence_sub_callback(void *pArg, int argc, char **argv, char **columnNames);
 
+
+
 struct presence_helper {
        sofia_profile_t *profile;
        switch_event_t *event;
@@ -186,7 +188,7 @@ char *sofia_presence_translate_rpid(char *in, char *ext)
        return r;
 }
 
-void sofia_presence_mwi_event_handler(switch_event_t *event)
+static void actual_sofia_presence_mwi_event_handler(switch_event_t *event)
 {
        char *account, *dup_account, *yn, *host, *user;
        char *sql;
@@ -275,7 +277,7 @@ void sofia_presence_mwi_event_handler(switch_event_t *event)
        }
 }
 
-void sofia_presence_event_handler(switch_event_t *event)
+static void actual_sofia_presence_event_handler(switch_event_t *event)
 {
        sofia_profile_t *profile = NULL;
        switch_hash_index_t *hi;
@@ -289,6 +291,11 @@ void sofia_presence_event_handler(switch_event_t *event)
        char *sql = NULL;
        char *euser = NULL, *user = NULL, *host = NULL;
 
+       
+       if (!mod_sofia_globals.running) {
+               return;
+       }
+       
        if (rpid && !strcasecmp(rpid, "n/a")) {
                rpid = NULL;
        }
@@ -452,19 +459,23 @@ void sofia_presence_event_handler(switch_event_t *event)
                        helper.profile = profile;
             helper.event = event;
                        SWITCH_STANDARD_STREAM(helper.stream);
+                       switch_assert(helper.stream.data);
+
                        sofia_glue_execute_sql_callback(profile,
                                                                                        SWITCH_FALSE,
-                                                                                       profile->ireg_mutex,
+                                                                                       //profile->ireg_mutex,
+                                                                                       NULL,
                                                                                        sql,
                                                                                        sofia_presence_sub_callback,
                                                                                        &helper);
                        
-
-                       if (!switch_strlen_zero((char *)helper.stream.data)) {
+                       if (switch_strlen_zero((char *)helper.stream.data)) {
+                               switch_safe_free(helper.stream.data);
+                       } else {
                                char *ssql = (char *)helper.stream.data;
                                sofia_glue_execute_sql(profile, &ssql, SWITCH_TRUE);
-                               helper.stream.data = NULL;
                        }
+                       helper.stream.data = NULL;
                }
        }
        switch_mutex_unlock(mod_sofia_globals.hash_mutex);
@@ -474,6 +485,99 @@ done:
        switch_safe_free(user);
 }
 
+
+void *SWITCH_THREAD_FUNC sofia_presence_event_thread_run(switch_thread_t *thread, void *obj)
+{
+       void *pop;
+       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Event Thread Started\n");
+
+       switch_mutex_lock(mod_sofia_globals.mutex);
+       mod_sofia_globals.threads++;
+       switch_mutex_unlock(mod_sofia_globals.mutex);
+
+       while (mod_sofia_globals.running == 1) {
+               int count = 0;
+               
+               if (switch_queue_trypop(mod_sofia_globals.presence_queue, &pop) == SWITCH_STATUS_SUCCESS) {
+                       switch_event_t *event = (switch_event_t *) pop;
+
+                       if (!pop) {
+                               break;
+                       }
+
+                       actual_sofia_presence_event_handler(event);
+                       switch_event_destroy(&event);
+                       count++;
+               }
+
+               if (switch_queue_trypop(mod_sofia_globals.mwi_queue, &pop) == SWITCH_STATUS_SUCCESS) {
+                       switch_event_t *event = (switch_event_t *) pop;
+
+                       if (!pop) {
+                               break;
+                       }
+
+                       actual_sofia_presence_mwi_event_handler(event);
+                       switch_event_destroy(&event);
+                       count++;
+               }
+
+               if (!count) {
+                       switch_yield(100000);
+               }
+       }
+
+       while (switch_queue_trypop(mod_sofia_globals.presence_queue, &pop) == SWITCH_STATUS_SUCCESS && pop) {
+               switch_event_t *event = (switch_event_t *) pop;
+               switch_event_destroy(&event);
+       }
+       
+       while (switch_queue_trypop(mod_sofia_globals.mwi_queue, &pop) == SWITCH_STATUS_SUCCESS && pop) {
+               switch_event_t *event = (switch_event_t *) pop;
+               switch_event_destroy(&event);
+       }
+
+       switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CONSOLE, "Event Thread Ended\n");
+
+       switch_mutex_lock(mod_sofia_globals.mutex);
+       mod_sofia_globals.threads--;
+       switch_mutex_unlock(mod_sofia_globals.mutex);
+
+       return NULL;
+}
+
+void sofia_presence_event_thread_start(void)
+{
+       switch_thread_t *thread;
+       switch_threadattr_t *thd_attr = NULL;
+
+       switch_threadattr_create(&thd_attr, mod_sofia_globals.pool);
+       switch_threadattr_detach_set(thd_attr, 1);
+       switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
+       switch_threadattr_priority_increase(thd_attr);
+       switch_thread_create(&thread, thd_attr, sofia_presence_event_thread_run, NULL, mod_sofia_globals.pool);
+}
+
+
+void sofia_presence_event_handler(switch_event_t *event)
+{
+       switch_event_t *cloned_event;
+
+       switch_event_dup(&cloned_event, event);
+       switch_assert(cloned_event);
+       switch_queue_push(mod_sofia_globals.presence_queue, cloned_event);
+}
+
+void sofia_presence_mwi_event_handler(switch_event_t *event)
+{
+       switch_event_t *cloned_event;
+
+       switch_event_dup(&cloned_event, event);
+       switch_assert(cloned_event);
+       switch_queue_push(mod_sofia_globals.mwi_queue, cloned_event);
+}
+
+
 static int sofia_presence_sub_reg_callback(void *pArg, int argc, char **argv, char **columnNames)
 {
        sofia_profile_t *profile = (sofia_profile_t *) pArg;