push events into a separate thread for processing

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

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.profile_hash, mod_sofia_globals.pool);
-	switch_core_hash_init(&mod_sofia_globals.gateway_hash, module_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, module_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;
 }

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);

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;
 	}
 

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;