FS-11848 [core] add user_data to event_channel_bind

* also adds a convenience switch_event_channel_deliver for already in thread

src/include/switch_event.h

@@ -428,8 +428,9 @@ SWITCH_DECLARE(void) switch_json_add_presence_data_cols(switch_event_t *event, c
 SWITCH_DECLARE(void) switch_event_launch_dispatch_threads(uint32_t max);
 
 SWITCH_DECLARE(switch_status_t) switch_event_channel_broadcast(const char *event_channel, cJSON **json, const char *key, switch_event_channel_id_t id);
-SWITCH_DECLARE(uint32_t) switch_event_channel_unbind(const char *event_channel, switch_event_channel_func_t func);
-SWITCH_DECLARE(switch_status_t) switch_event_channel_bind(const char *event_channel, switch_event_channel_func_t func, switch_event_channel_id_t *id);
+SWITCH_DECLARE(switch_status_t) switch_event_channel_deliver(const char *event_channel, cJSON **json, const char *key, switch_event_channel_id_t id);
+SWITCH_DECLARE(uint32_t) switch_event_channel_unbind(const char *event_channel, switch_event_channel_func_t func, void *user_data);
+SWITCH_DECLARE(switch_status_t) switch_event_channel_bind(const char *event_channel, switch_event_channel_func_t func, switch_event_channel_id_t *id, void *user_data);
 
 
 typedef void (*switch_live_array_command_handler_t)(switch_live_array_t *la, const char *cmd, const char *sessid, cJSON *jla, void *user_data);

src/include/switch_types.h

@@ -2621,7 +2621,7 @@ struct switch_media_handle_s;
 typedef struct switch_media_handle_s switch_media_handle_t;
 
 typedef uint32_t switch_event_channel_id_t;
-typedef void (*switch_event_channel_func_t)(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id);
+typedef void (*switch_event_channel_func_t)(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data);
 
 struct switch_live_array_s;
 typedef struct switch_live_array_s switch_live_array_t;

src/mod/applications/mod_conference/conference_event.c

@@ -57,7 +57,8 @@ static cJSON *get_canvas_info(mcu_canvas_t *canvas)
 	return obj;
 }
 
-void conference_event_mod_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id)
+
+void conference_event_mod_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data)
 {
 	cJSON *data, *addobj = NULL;
 	const char *action = NULL;
@@ -413,7 +414,7 @@ void conference_event_mod_channel_handler(const char *event_channel, cJSON *json
 
 }
 
-void conference_event_chat_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id)
+void conference_event_chat_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data)
 {
 	cJSON *data;
 	cJSON *jid = 0;
@@ -475,12 +476,12 @@ void conference_event_chat_channel_handler(const char *event_channel, cJSON *jso
 	switch_safe_free(conference_name);
 }
 
-void conference_event_la_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id)
+void conference_event_la_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data)
 {
 	switch_live_array_parse_json(json, conference_globals.event_channel_id);
 }
 
-void conference_event_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id)
+void conference_event_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data)
 {
 	char *domain = NULL, *name = NULL;
 	conference_obj_t *conference = NULL;

src/mod/applications/mod_conference/mod_conference.c

@@ -3934,10 +3934,10 @@ SWITCH_MODULE_LOAD_FUNCTION(mod_conference_load)
 	switch_console_add_complete_func("::conference::conference_list_conferences", conference_list_conferences);
 
 
-	switch_event_channel_bind("conference", conference_event_channel_handler, &conference_globals.event_channel_id);
-	switch_event_channel_bind("conference-liveArray", conference_event_la_channel_handler, &conference_globals.event_channel_id);
-	switch_event_channel_bind("conference-mod", conference_event_mod_channel_handler, &conference_globals.event_channel_id);
-	switch_event_channel_bind("conference-chat", conference_event_chat_channel_handler, &conference_globals.event_channel_id);
+	switch_event_channel_bind("conference", conference_event_channel_handler, &conference_globals.event_channel_id, NULL);
+	switch_event_channel_bind("conference-liveArray", conference_event_la_channel_handler, &conference_globals.event_channel_id, NULL);
+	switch_event_channel_bind("conference-mod", conference_event_mod_channel_handler, &conference_globals.event_channel_id, NULL);
+	switch_event_channel_bind("conference-chat", conference_event_chat_channel_handler, &conference_globals.event_channel_id, NULL);
 
 	if ( conference_api_sub_syntax(&api_syntax) != SWITCH_STATUS_SUCCESS) {
 		return SWITCH_STATUS_TERM;
@@ -3991,10 +3991,10 @@ SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_conference_shutdown)
 		/* signal all threads to shutdown */
 		conference_globals.running = 0;
 
-		switch_event_channel_unbind(NULL, conference_event_channel_handler);
-		switch_event_channel_unbind(NULL, conference_event_la_channel_handler);
-		switch_event_channel_unbind(NULL, conference_event_mod_channel_handler);
-		switch_event_channel_unbind(NULL, conference_event_chat_channel_handler);
+		switch_event_channel_unbind(NULL, conference_event_channel_handler, NULL);
+		switch_event_channel_unbind(NULL, conference_event_la_channel_handler, NULL);
+		switch_event_channel_unbind(NULL, conference_event_mod_channel_handler, NULL);
+		switch_event_channel_unbind(NULL, conference_event_chat_channel_handler, NULL);
 
 		switch_console_del_complete_func("::conference::conference_list_conferences");
 

src/mod/applications/mod_conference/mod_conference.h

@@ -1167,10 +1167,10 @@ void conference_cdr_del(conference_member_t *member);
 void conference_cdr_add(conference_member_t *member);
 void conference_cdr_rejected(conference_obj_t *conference, switch_channel_t *channel, cdr_reject_reason_t reason);
 void conference_cdr_render(conference_obj_t *conference);
-void conference_event_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id);
-void conference_event_la_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id);
-void conference_event_mod_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id);
-void conference_event_chat_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id);
+void conference_event_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data);
+void conference_event_la_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data);
+void conference_event_mod_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data);
+void conference_event_chat_channel_handler(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data);
 
 void conference_member_itterator(conference_obj_t *conference, switch_stream_handle_t *stream, uint8_t non_mod, conference_api_member_cmd_t pfncallback, void *data);
 int conference_video_flush_queue(switch_queue_t *q, int min);

src/mod/endpoints/mod_verto/mod_verto.c

@@ -171,7 +171,7 @@ static void close_socket(ws_socket_t *sock)
 	}
 }
 
-void verto_broadcast(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id);
+void verto_broadcast(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data);
 
 static int verto_init_ssl(verto_profile_t *profile)
 {
@@ -5570,7 +5570,7 @@ static switch_call_cause_t verto_outgoing_channel(switch_core_session_t *session
 	return cause;
 }
 
-void verto_broadcast(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id)
+void verto_broadcast(const char *event_channel, cJSON *json, const char *key, switch_event_channel_id_t id, void *user_data)
 {
 	if (verto_globals.debug > 9) {
 		char *json_text;
@@ -6113,7 +6113,7 @@ static void event_handler(switch_event_t *event)
 
 	/* comment broadcasting globally and change to only within the module cos FS events are heavy */
 	//switch_event_channel_broadcast(event_channel, &msg, __FILE__, NO_EVENT_CHANNEL_ID);
-	verto_broadcast(event_channel, msg, __FILE__, NO_EVENT_CHANNEL_ID);cJSON_Delete(msg);
+	verto_broadcast(event_channel, msg, __FILE__, NO_EVENT_CHANNEL_ID, NULL);cJSON_Delete(msg);
 
 	free(event_channel);
 
@@ -6185,7 +6185,7 @@ SWITCH_MODULE_LOAD_FUNCTION(mod_verto_load)
 
 
 
-	switch_event_channel_bind(SWITCH_EVENT_CHANNEL_GLOBAL, verto_broadcast, &verto_globals.event_channel_id);
+	switch_event_channel_bind(SWITCH_EVENT_CHANNEL_GLOBAL, verto_broadcast, &verto_globals.event_channel_id, NULL);
 
 
 	r = init();
@@ -6243,7 +6243,7 @@ SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_verto_shutdown)
 	json_cleanup();
 	switch_core_hash_destroy(&json_GLOBALS.store_hash);
 
-	switch_event_channel_unbind(NULL, verto_broadcast);
+	switch_event_channel_unbind(NULL, verto_broadcast, NULL);
 	switch_event_unbind_callback(presence_event_handler);
 	switch_event_unbind_callback(event_handler);
 

src/switch_event.c

@@ -2698,9 +2698,9 @@ SWITCH_DECLARE(void) switch_event_add_presence_data_cols(switch_channel_t *chann
 }
 
 struct switch_event_channel_sub_node_head_s;
-
 typedef struct switch_event_channel_sub_node_s {
 	switch_event_channel_func_t func;
+	void *user_data;
 	switch_event_channel_id_t id;
 	struct switch_event_channel_sub_node_head_s *head;
 	struct switch_event_channel_sub_node_s *next;
@@ -2712,7 +2712,7 @@ typedef struct switch_event_channel_sub_node_head_s {
 	char *event_channel;
 } switch_event_channel_sub_node_head_t;
 
-static uint32_t switch_event_channel_unsub_head(switch_event_channel_func_t func, switch_event_channel_sub_node_head_t *head)
+static uint32_t switch_event_channel_unsub_head(switch_event_channel_func_t func, switch_event_channel_sub_node_head_t *head, void *user_data)
 {
 	uint32_t x = 0;
 
@@ -2725,7 +2725,7 @@ static uint32_t switch_event_channel_unsub_head(switch_event_channel_func_t func
 		thisnp = np;
 		np = np->next;
 
-		if (!func || thisnp->func == func) {
+		if (!(func) || (thisnp->func == func && (thisnp->user_data == user_data || user_data == NULL))) {
 			x++;
 
 			if (last) {
@@ -2769,7 +2769,7 @@ static void unsub_all_switch_event_channel(void)
 	while ((hi = switch_core_hash_first_iter( event_channel_manager.hash, hi))) {
 		switch_core_hash_this(hi, NULL, NULL, &val);
 		head = (switch_event_channel_sub_node_head_t *) val;
-		switch_event_channel_unsub_head(NULL, head);
+		switch_event_channel_unsub_head(NULL, head, NULL);
 		switch_core_hash_delete(event_channel_manager.hash, head->event_channel);
 		free(head->event_channel);
 		free(head);
@@ -2779,7 +2779,7 @@ static void unsub_all_switch_event_channel(void)
 	switch_thread_rwlock_unlock(event_channel_manager.rwlock);
 }
 
-static uint32_t switch_event_channel_unsub_channel(switch_event_channel_func_t func, const char *event_channel)
+static uint32_t switch_event_channel_unsub_channel(switch_event_channel_func_t func, const char *event_channel, void *user_data)
 {
 	switch_event_channel_sub_node_head_t *head;
 	uint32_t x = 0;
@@ -2795,13 +2795,13 @@ static uint32_t switch_event_channel_unsub_channel(switch_event_channel_func_t f
 
 			if (val) {
 				head = (switch_event_channel_sub_node_head_t *) val;
-				x += switch_event_channel_unsub_head(func, head);
+				x += switch_event_channel_unsub_head(func, head, user_data);
 			}
 		}
 
 	} else {
 		if ((head = switch_core_hash_find(event_channel_manager.hash, event_channel))) {
-			x += switch_event_channel_unsub_head(func, head);
+			x += switch_event_channel_unsub_head(func, head, user_data);
 		}
 	}
 
@@ -2810,7 +2810,7 @@ static uint32_t switch_event_channel_unsub_channel(switch_event_channel_func_t f
 	return x;
 }
 
-static switch_status_t switch_event_channel_sub_channel(const char *event_channel, switch_event_channel_func_t func, switch_event_channel_id_t id)
+static switch_status_t switch_event_channel_sub_channel(const char *event_channel, switch_event_channel_func_t func, switch_event_channel_id_t id, void *user_data)
 
 {
 	switch_event_channel_sub_node_t *node, *np;
@@ -2826,6 +2826,7 @@ static switch_status_t switch_event_channel_sub_channel(const char *event_channe
 
 		switch_zmalloc(node, sizeof(*node));
 		node->func = func;
+		node->user_data = user_data;
 		node->id = id;
 
 		node->head = head;
@@ -2836,7 +2837,7 @@ static switch_status_t switch_event_channel_sub_channel(const char *event_channe
 		int exist = 0;
 
 		for (np = head->node; np; np = np->next) {
-			if (np->func == func) {
+			if (np->func == func && np->user_data == user_data) {
 				exist = 1;
 				break;
 			}
@@ -2846,6 +2847,7 @@ static switch_status_t switch_event_channel_sub_channel(const char *event_channe
 			switch_zmalloc(node, sizeof(*node));
 
 			node->func = func;
+			node->user_data = user_data;
 			node->id = id;
 			node->head = head;
 
@@ -2889,7 +2891,7 @@ static uint32_t _switch_event_channel_broadcast(const char *event_channel, const
 				continue;
 			}
 
-			np->func(broadcast_channel, json, key, id);
+			np->func(broadcast_channel, json, key, id, np->user_data);
 			x++;
 		}
 	}
@@ -3034,13 +3036,29 @@ SWITCH_DECLARE(switch_status_t) switch_event_channel_broadcast(const char *event
 	return status;
 }
 
-SWITCH_DECLARE(uint32_t) switch_event_channel_unbind(const char *event_channel, switch_event_channel_func_t func)
+SWITCH_DECLARE(switch_status_t) switch_event_channel_deliver(const char *event_channel, cJSON **json, const char *key, switch_event_channel_id_t id)
 {
-	return switch_event_channel_unsub_channel(func, event_channel);
+	event_channel_data_t *ecd = NULL;
+	switch_zmalloc(ecd, sizeof(*ecd));
+
+	ecd->event_channel = strdup(event_channel);
+	ecd->json = *json;
+	ecd->key = strdup(key);
+	ecd->id = id;
+
+	*json = NULL;
+
+	ecd_deliver(&ecd);
+
+	return SWITCH_STATUS_SUCCESS;
 }
 
-SWITCH_DECLARE(switch_status_t) switch_event_channel_bind(const char *event_channel, switch_event_channel_func_t func, switch_event_channel_id_t *id)
+SWITCH_DECLARE(uint32_t) switch_event_channel_unbind(const char *event_channel, switch_event_channel_func_t func, void *user_data)
+{
+	return switch_event_channel_unsub_channel(func, event_channel, user_data);
+}
 
+SWITCH_DECLARE(switch_status_t) switch_event_channel_bind(const char *event_channel, switch_event_channel_func_t func, switch_event_channel_id_t *id, void *user_data)
 {
 	switch_status_t status = SWITCH_STATUS_SUCCESS;
 
@@ -3052,7 +3070,7 @@ SWITCH_DECLARE(switch_status_t) switch_event_channel_bind(const char *event_chan
 		switch_thread_rwlock_unlock(event_channel_manager.rwlock);
 	}
 
-	status = switch_event_channel_sub_channel(event_channel, func, *id);
+	status = switch_event_channel_sub_channel(event_channel, func, *id, user_data);
 
 	return status;
 }