[asterisk-commits] kmoore: branch kmoore/stasis-channel_events-take2 r384487 - /team/kmoore/stas...

[SVN commits to the Asterisk project]

Author: kmoore
Date: Mon Apr  1 14:59:32 2013
New Revision: 384487

URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=384487
Log:
Move channel message handling into a message router and refactor accordingly.  Add channel blob message support.

Modified:
    team/kmoore/stasis-channel_events-take2/apps/app_stasis.c

Modified: team/kmoore/stasis-channel_events-take2/apps/app_stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/kmoore/stasis-channel_events-take2/apps/app_stasis.c?view=diff&rev=384487&r1=384486&r2=384487
==============================================================================
--- team/kmoore/stasis-channel_events-take2/apps/app_stasis.c (original)
+++ team/kmoore/stasis-channel_events-take2/apps/app_stasis.c Mon Apr  1 14:59:32 2013
@@ -37,6 +37,7 @@
 #include "asterisk/channel.h"
 #include "asterisk/module.h"
 #include "asterisk/stasis.h"
+#include "asterisk/stasis_message_router.h"
 #include "asterisk/strings.h"
 
 /*** DOCUMENTATION
@@ -322,22 +323,45 @@
 	return 0;
 }
 
-static void sub_handler(void *data, struct stasis_subscription *sub,
-			struct stasis_topic *topic,
-			struct stasis_message *message)
-{
+static void sub_snapshot_handler(void *data,
+		struct stasis_subscription *sub,
+		struct stasis_topic *topic,
+		struct stasis_message *message)
+{
+	RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
+	struct ast_channel_snapshot *snapshot = stasis_message_data(message);
 	struct app *app = data;
-	if (ast_channel_snapshot_type() == stasis_message_type(message)) {
-		RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
-		struct ast_channel_snapshot *snapshot =
-			stasis_message_data(message);
-
-		msg = app_event_create("channel-state-change", snapshot, NULL);
-		if (!msg) {
-			return;
-		}
-		app_send(app, msg);
-	}
+
+	msg = app_event_create("channel-state-change", snapshot, NULL);
+	if (!msg) {
+		return;
+	}
+
+	app_send(app, msg);
+}
+
+static void sub_blob_handler(void *data,
+		struct stasis_subscription *sub,
+		struct stasis_topic *topic,
+		struct stasis_message *message)
+{
+	RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
+	struct ast_channel_blob *obj = stasis_message_data(message);
+	struct app *app = data;
+
+	msg = app_event_create("channel-event", obj->snapshot, obj->blob);
+	if (!msg) {
+		return;
+	}
+
+	app_send(app, msg);
+}
+
+static void sub_change_handler(void *data,
+		struct stasis_subscription *sub,
+		struct stasis_topic *topic,
+		struct stasis_message *message)
+{
 	if (stasis_subscription_final_message(sub, message)) {
 		ao2_cleanup(data);
 	}
@@ -366,7 +390,7 @@
 	RAII_VAR(struct ao2_container *, apps, apps_registry(), ao2_cleanup);
 	RAII_VAR(struct app *, app, NULL, ao2_cleanup);
 	RAII_VAR(struct stasis_app_control *, control, NULL, control_unlink);
-	RAII_VAR(struct stasis_subscription *, subscription, NULL, stasis_unsubscribe);
+	RAII_VAR(struct stasis_message_router *, router, NULL, stasis_message_router_unsubscribe);
 	int res = 0;
 	char *parse = NULL;
 	int hungup = 0;
@@ -406,12 +430,24 @@
 		ao2_link(controls, control);
 	}
 
-	subscription = stasis_subscribe(ast_channel_topic(chan), sub_handler, app);
-	if (subscription == NULL) {
-		ast_log(LOG_ERROR, "Error subscribing app %s to channel %s\n", args.app_name, ast_channel_name(chan));
-		return -1;
-	}
-	ao2_ref(app, +1); /* subscription now has a reference */
+	router = stasis_message_router_create(ast_channel_topic(chan));
+	if (!router) {
+		ast_log(LOG_ERROR, "Error creating message router for app %s to channel %s\n", args.app_name, ast_channel_name(chan));
+		return -1;
+	}
+
+	res |= stasis_message_router_add(router, stasis_subscription_change_type(), sub_change_handler, app);
+	if (!res) {
+		ao2_ref(app, +1); /* subscription now has a reference */
+	}
+
+	res |= stasis_message_router_add(router, ast_channel_snapshot_type(), sub_snapshot_handler, app);
+	res |= stasis_message_router_add(router, ast_channel_blob_type(), sub_blob_handler, app);
+
+	if (res) {
+		ast_log(LOG_ERROR, "Error adding routes for app %s to channel %s\n", args.app_name, ast_channel_name(chan));
+		return -1;
+	}
 
 	res = send_start_msg(app, chan, args.argc - 1, args.app_argv);
 	if (res != 0) {