Remove required type field from channel blobs

When we first introduced the channel blob types, the JSON blobs were
self identifying by a required "type" field in the JSON object
itself. This, as it turns out, was a bad idea.

When we introduced the message router, it was useless for routing based
on the JSON type. And messages had two type fields to check: the
stasis_message_type() of the message itself, plus the type field in the
JSON blob (but only if it was a blob message).

This patch corrects that mistake by removing the required type field
from JSON blobs, and introducing first class stasis_message_type objects
for the actual message type.

Since we now will have a proliferation of message types, I introduced a
few macros to help reduce the amount of boilerplate necessary to set
them up.

Review: https://reviewboard.asterisk.org/r/2509


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@388005 65c4cc65-6c06-0410-ace0-fbb531ad65f3

3 files changed, 83 insertions, 120 deletions

diff --git a/main/channel.c b/main/channel.c
index 9ed7e5f25..d63aeb2ba 100644
--- a/main/channel.c
+++ b/main/channel.c
@@ -1368,11 +1368,12 @@ int ast_queue_frame_head(struct ast_channel *chan, struct ast_frame *fin)
 }
 
 /*! \internal \brief Publish a channel blob message */
-static void publish_channel_blob(struct ast_channel *chan, struct ast_json *blob)
+static void publish_channel_blob(struct ast_channel *chan,
+	struct stasis_message_type *type, struct ast_json *blob)
 {
 	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
 	if (blob) {
-		message = ast_channel_blob_create(chan, blob);
+		message = ast_channel_blob_create(chan, type, blob);
 	}
 	if (message) {
 		stasis_publish(ast_channel_topic(chan), message);
@@ -1382,7 +1383,6 @@ static void publish_channel_blob(struct ast_channel *chan, struct ast_json *blob
 /*! \brief Queue a hangup frame for channel */
 int ast_queue_hangup(struct ast_channel *chan)
 {
-	RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
 	struct ast_frame f = { AST_FRAME_CONTROL, .subclass.integer = AST_CONTROL_HANGUP };
 	int res;
@@ -1390,8 +1390,7 @@ int ast_queue_hangup(struct ast_channel *chan)
 	/* Yeah, let's not change a lock-critical value without locking */
 	ast_channel_lock(chan);
 	ast_channel_softhangup_internal_flag_add(chan, AST_SOFTHANGUP_DEV);
-	blob = ast_json_pack("{s: s}", "type", "hangup_request");
-	publish_channel_blob(chan, blob);
+	publish_channel_blob(chan, ast_channel_hangup_request_type(), NULL);
 
 	res = ast_queue_frame(chan, &f);
 	ast_channel_unlock(chan);
@@ -1416,10 +1415,9 @@ int ast_queue_hangup_with_cause(struct ast_channel *chan, int cause)
 	if (cause < 0) {
 		f.data.uint32 = ast_channel_hangupcause(chan);
 	}
-	blob = ast_json_pack("{s: s, s: i}",
-			     "type", "hangup_request",
+	blob = ast_json_pack("{s: i}",
 			     "cause", cause);
-	publish_channel_blob(chan, blob);
+	publish_channel_blob(chan, ast_channel_hangup_request_type(), blob);
 
 	res = ast_queue_frame(chan, &f);
 	ast_channel_unlock(chan);
@@ -2727,11 +2725,10 @@ int ast_softhangup(struct ast_channel *chan, int cause)
 
 	ast_channel_lock(chan);
 	res = ast_softhangup_nolock(chan, cause);
-	blob = ast_json_pack("{s: s, s: i, s: b}",
-			     "type", "hangup_request",
+	blob = ast_json_pack("{s: i, s: b}",
 			     "cause", cause,
 			     "soft", 1);
-	publish_channel_blob(chan, blob);
+	publish_channel_blob(chan, ast_channel_hangup_request_type(), blob);
 	ast_channel_unlock(chan);
 
 	return res;
@@ -3737,15 +3734,14 @@ static void send_dtmf_begin_event(struct ast_channel *chan,
 	RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 	char digit_str[] = { digit, '\0' };
 
-	blob = ast_json_pack("{ s: s, s: s, s: s }",
-		"type", "dtmf_begin",
+	blob = ast_json_pack("{ s: s, s: s }",
 		"digit", digit_str,
 		"direction", dtmf_direction_to_string(direction));
 	if (!blob) {
 		return;
 	}
 
-	publish_channel_blob(chan, blob);
+	publish_channel_blob(chan, ast_channel_dtmf_begin_type(), blob);
 }
 
 static void send_dtmf_end_event(struct ast_channel *chan,
@@ -3754,8 +3750,7 @@ static void send_dtmf_end_event(struct ast_channel *chan,
 	RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 	char digit_str[] = { digit, '\0' };
 
-	blob = ast_json_pack("{ s: s, s: s, s: s, s: i }",
-		"type", "dtmf_end",
+	blob = ast_json_pack("{ s: s, s: s, s: i }",
 		"digit", digit_str,
 		"direction", dtmf_direction_to_string(direction),
 		"duration_ms", duration_ms);
@@ -3763,7 +3758,7 @@ static void send_dtmf_end_event(struct ast_channel *chan,
 		return;
 	}
 
-	publish_channel_blob(chan, blob);
+	publish_channel_blob(chan, ast_channel_dtmf_end_type(), blob);
 }
 
 static void ast_read_generator_actions(struct ast_channel *chan, struct ast_frame *f)
diff --git a/main/manager_channels.c b/main/manager_channels.c
index 004467e3b..804bca663 100644
--- a/main/manager_channels.c
+++ b/main/manager_channels.c
@@ -544,8 +544,10 @@ static void channel_snapshot_update(void *data, struct stasis_subscription *sub,
 	}
 }
 
-static void channel_varset(struct ast_channel_blob *obj)
+static void channel_varset_cb(void *data, struct stasis_subscription *sub,
+	struct stasis_topic *topic, struct stasis_message *message)
 {
+	struct ast_channel_blob *obj = stasis_message_data(message);
 	RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
 	const char *variable = ast_json_string_get(ast_json_object_get(obj->blob, "variable"));
 	const char *value = ast_json_string_get(ast_json_object_get(obj->blob, "value"));
@@ -585,8 +587,10 @@ static void channel_varset(struct ast_channel_blob *obj)
 		      variable, value);
 }
 
-static void channel_userevent(struct ast_channel_blob *obj)
+static void channel_user_event_cb(void *data, struct stasis_subscription *sub,
+	struct stasis_topic *topic, struct stasis_message *message)
 {
+	struct ast_channel_blob *obj = stasis_message_data(message);
 	RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
 	const char *eventname;
 	const char *body;
@@ -620,8 +624,11 @@ static void channel_userevent(struct ast_channel_blob *obj)
 		      ast_str_buffer(channel_event_string), eventname, body);
 }
 
-static void channel_hangup_request(struct ast_channel_blob *obj)
+static void channel_hangup_request_cb(void *data,
+	struct stasis_subscription *sub, struct stasis_topic *topic,
+	struct stasis_message *message)
 {
+	struct ast_channel_blob *obj = stasis_message_data(message);
 	RAII_VAR(struct ast_str *, extra, NULL, ast_free);
 	RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
 	struct ast_json *cause;
@@ -657,8 +664,10 @@ static void channel_hangup_request(struct ast_channel_blob *obj)
 		      ast_str_buffer(extra));
 }
 
-static void channel_dtmf_begin(struct ast_channel_blob *obj)
+static void channel_dtmf_begin_cb(void *data, struct stasis_subscription *sub,
+	struct stasis_topic *topic, struct stasis_message *message)
 {
+	struct ast_channel_blob *obj = stasis_message_data(message);
 	RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
 	const char *digit =
 		ast_json_string_get(ast_json_object_get(obj->blob, "digit"));
@@ -696,8 +705,10 @@ static void channel_dtmf_begin(struct ast_channel_blob *obj)
 		digit, direction);
 }
 
-static void channel_dtmf_end(struct ast_channel_blob *obj)
+static void channel_dtmf_end_cb(void *data, struct stasis_subscription *sub,
+	struct stasis_topic *topic, struct stasis_message *message)
 {
+	struct ast_channel_blob *obj = stasis_message_data(message);
 	RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
 	const char *digit =
 		ast_json_string_get(ast_json_object_get(obj->blob, "digit"));
@@ -742,33 +753,10 @@ static void channel_dtmf_end(struct ast_channel_blob *obj)
 }
 
 /*!
- * \brief Callback processing messages on the channel topic.
- */
-static void channel_blob_cb(void *data, struct stasis_subscription *sub,
-			    struct stasis_topic *topic,
-			    struct stasis_message *message)
-{
-	struct ast_channel_blob *obj = stasis_message_data(message);
-
-	if (strcmp("varset", ast_channel_blob_json_type(obj)) == 0) {
-		channel_varset(obj);
-	} else if (strcmp("userevent", ast_channel_blob_json_type(obj)) == 0) {
-		channel_userevent(obj);
-	} else if (strcmp("hangup_request", ast_channel_blob_json_type(obj)) == 0) {
-		channel_hangup_request(obj);
-	} else if (strcmp("dtmf_begin", ast_channel_blob_json_type(obj)) == 0) {
-		channel_dtmf_begin(obj);
-	} else if (strcmp("dtmf_end", ast_channel_blob_json_type(obj)) == 0) {
-		channel_dtmf_end(obj);
-	}
-}
-
-/*!
  * \brief Callback processing messages for channel dialing
  */
 static void channel_dial_cb(void *data, struct stasis_subscription *sub,
-				struct stasis_topic *topic,
-				struct stasis_message *message)
+	struct stasis_topic *topic, struct stasis_message *message)
 {
 	struct ast_multi_channel_blob *obj = stasis_message_data(message);
 	const char *dialstatus;
@@ -778,11 +766,6 @@ static void channel_dial_cb(void *data, struct stasis_subscription *sub,
 	RAII_VAR(struct ast_str *, caller_event_string, NULL, ast_free);
 	RAII_VAR(struct ast_str *, peer_event_string, NULL, ast_free);
 
-	if (strcmp("dial", ast_multi_channel_blob_get_type(obj))) {
-		ast_assert(0);
-		return;
-	}
-
 	caller = ast_multi_channel_blob_get_channel(obj, "caller");
 	peer = ast_multi_channel_blob_get_channel(obj, "peer");
 
@@ -852,8 +835,28 @@ int manager_channels_init(void)
 					 NULL);
 
 	ret |= stasis_message_router_add(channel_state_router,
-					 ast_channel_blob_type(),
-					 channel_blob_cb,
+					 ast_channel_varset_type(),
+					 channel_varset_cb,
+					 NULL);
+
+	ret |= stasis_message_router_add(channel_state_router,
+					 ast_channel_user_event_type(),
+					 channel_user_event_cb,
+					 NULL);
+
+	ret |= stasis_message_router_add(channel_state_router,
+					 ast_channel_dtmf_begin_type(),
+					 channel_dtmf_begin_cb,
+					 NULL);
+
+	ret |= stasis_message_router_add(channel_state_router,
+					 ast_channel_dtmf_end_type(),
+					 channel_dtmf_end_cb,
+					 NULL);
+
+	ret |= stasis_message_router_add(channel_state_router,
+					 ast_channel_hangup_request_type(),
+					 channel_hangup_request_cb,
 					 NULL);
 
 	ret |= stasis_message_router_add(channel_state_router,
diff --git a/main/stasis_channels.c b/main/stasis_channels.c
index b40f13572..c032ec956 100644
--- a/main/stasis_channels.c
+++ b/main/stasis_channels.c
@@ -38,14 +38,17 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 
 #define NUM_MULTI_CHANNEL_BLOB_BUCKETS 7
 
-/*! \brief Message type for channel snapshot messages */
-static struct stasis_message_type *channel_snapshot_type;
-
-/*! \brief Message type for channel blob messages */
-static struct stasis_message_type *channel_blob_type;
-
-/*! \brief Message type for channel dial messages */
-static struct stasis_message_type *channel_dial_type;
+/*!
+ * @{ \brief Define channel message types.
+ */
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_snapshot_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_dial_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_varset_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_user_event_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_hangup_request_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_dtmf_begin_type);
+STASIS_MESSAGE_TYPE_DEFN(ast_channel_dtmf_end_type);
+/*! @} */
 
 /*! \brief Topic for all channels */
 struct stasis_topic *channel_topic_all;
@@ -53,21 +56,6 @@ struct stasis_topic *channel_topic_all;
 /*! \brief Caching topic for all channels */
 struct stasis_caching_topic *channel_topic_all_cached;
 
-struct stasis_message_type *ast_channel_dial_type(void)
-{
-	return channel_dial_type;
-}
-
-struct stasis_message_type *ast_channel_blob_type(void)
-{
-	return channel_blob_type;
-}
-
-struct stasis_message_type *ast_channel_snapshot_type(void)
-{
-	return channel_snapshot_type;
-}
-
 struct stasis_topic *ast_channel_topic_all(void)
 {
 	return channel_topic_all;
@@ -221,18 +209,13 @@ void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *pe
 }
 
 struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
-					       struct ast_json *blob)
+	struct stasis_message_type *type, struct ast_json *blob)
 {
 	RAII_VAR(struct ast_channel_blob *, obj, NULL, ao2_cleanup);
 	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
-	struct ast_json *type;
 
-	ast_assert(blob != NULL);
-
-	type = ast_json_object_get(blob, "type");
-	if (type == NULL) {
-		ast_log(LOG_ERROR, "Invalid ast_channel_blob; missing type field\n");
-		return NULL;
+	if (blob == NULL) {
+		blob = ast_json_null();
 	}
 
 	obj = ao2_alloc(sizeof(*obj), channel_blob_dtor);
@@ -249,7 +232,7 @@ struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
 
 	obj->blob = ast_json_ref(blob);
 
-	msg = stasis_message_create(ast_channel_blob_type(), obj);
+	msg = stasis_message_create(type, obj);
 	if (!msg) {
 		return NULL;
 	}
@@ -258,15 +241,6 @@ struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
 	return msg;
 }
 
-const char *ast_channel_blob_json_type(struct ast_channel_blob *obj)
-{
-	if (obj == NULL) {
-		return NULL;
-	}
-
-	return ast_json_string_get(ast_json_object_get(obj->blob, "type"));
-}
-
 /*! \brief A channel snapshot wrapper object used in \ref ast_multi_channel_blob objects */
 struct channel_role_snapshot {
 	struct ast_channel_snapshot *snapshot;	/*!< A channel snapshot */
@@ -319,7 +293,6 @@ struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *bl
 	RAII_VAR(struct ast_multi_channel_blob *, obj,
 			ao2_alloc(sizeof(*obj), multi_channel_blob_dtor),
 			ao2_cleanup);
-	struct ast_json *type;
 
 	ast_assert(blob != NULL);
 
@@ -327,12 +300,6 @@ struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *bl
 		return NULL;
 	}
 
-	type = ast_json_object_get(blob, "type");
-	if (type == NULL) {
-		ast_log(LOG_ERROR, "Invalid ast_multi_channel_blob; missing type field\n");
-		return NULL;
-	}
-
 	obj->channel_snapshots = ao2_container_alloc(NUM_MULTI_CHANNEL_BLOB_BUCKETS,
 			channel_role_hash_cb, channel_role_single_cmp_cb);
 	if (!obj->channel_snapshots) {
@@ -423,15 +390,6 @@ struct ast_json *ast_multi_channel_blob_get_json(struct ast_multi_channel_blob *
 	return obj->blob;
 }
 
-const char *ast_multi_channel_blob_get_type(struct ast_multi_channel_blob *obj)
-{
-	if (!obj) {
-		return NULL;
-	}
-
-	return ast_json_string_get(ast_json_object_get(obj->blob, "type"));
-}
-
 void ast_channel_publish_varset(struct ast_channel *chan, const char *name, const char *value)
 {
 	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
@@ -449,7 +407,8 @@ void ast_channel_publish_varset(struct ast_channel *chan, const char *name, cons
 		return;
 	}
 
-	msg = ast_channel_blob_create(chan, ast_json_ref(blob));
+	msg = ast_channel_blob_create(chan, ast_channel_varset_type(),
+		ast_json_ref(blob));
 
 	if (!msg) {
 		return;
@@ -491,12 +450,13 @@ struct ast_json *ast_channel_snapshot_to_json(const struct ast_channel_snapshot
 
 void ast_stasis_channels_shutdown(void)
 {
-	ao2_cleanup(channel_snapshot_type);
-	channel_snapshot_type = NULL;
-	ao2_cleanup(channel_blob_type);
-	channel_blob_type = NULL;
-	ao2_cleanup(channel_dial_type);
-	channel_dial_type = NULL;
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_snapshot_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dial_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_varset_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_user_event_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_hangup_request_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dtmf_begin_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dtmf_end_type);
 	ao2_cleanup(channel_topic_all);
 	channel_topic_all = NULL;
 	channel_topic_all_cached = stasis_caching_unsubscribe(channel_topic_all_cached);
@@ -504,9 +464,14 @@ void ast_stasis_channels_shutdown(void)
 
 void ast_stasis_channels_init(void)
 {
-	channel_snapshot_type = stasis_message_type_create("ast_channel_snapshot");
-	channel_blob_type = stasis_message_type_create("ast_channel_blob");
-	channel_dial_type = stasis_message_type_create("ast_channel_dial");
+	STASIS_MESSAGE_TYPE_INIT(ast_channel_snapshot_type);
+	STASIS_MESSAGE_TYPE_INIT(ast_channel_dial_type);
+	STASIS_MESSAGE_TYPE_INIT(ast_channel_varset_type);
+	STASIS_MESSAGE_TYPE_INIT(ast_channel_user_event_type);
+	STASIS_MESSAGE_TYPE_INIT(ast_channel_hangup_request_type);
+	STASIS_MESSAGE_TYPE_INIT(ast_channel_dtmf_begin_type);
+	STASIS_MESSAGE_TYPE_INIT(ast_channel_dtmf_end_type);
+
 	channel_topic_all = stasis_topic_create("ast_channel_topic_all");
 	channel_topic_all_cached = stasis_caching_topic_create(channel_topic_all, channel_snapshot_get_id);
 }