CEL refactoring cleanup

This change removes AST_CEL_BRIDGE_UPDATE since it should no longer be
used because masquerade situations are now accounted for in other ways.

This also refactors usage of AST_CEL_FORWARD to be produced by a Dial
message which has been extended with a "forward" field.

(closes issue ASTERISK-21566)
Review: https://reviewboard.asterisk.org/r/2635/


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

apps/app_celgenuserevent.c

@@ -62,6 +62,7 @@ static int celgenuserevent_exec(struct ast_channel *chan, const char *data)
 {
 	int res = 0;
 	char *parse;
+	RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
 	AST_DECLARE_APP_ARGS(args,
 		AST_APP_ARG(event);
 		AST_APP_ARG(extra);
@@ -74,7 +75,13 @@ static int celgenuserevent_exec(struct ast_channel *chan, const char *data)
 	parse = ast_strdupa(data);
 	AST_STANDARD_APP_ARGS(args, parse);
 
-	ast_cel_report_event(chan, AST_CEL_USER_DEFINED, args.event, args.extra, NULL);
+	blob = ast_json_pack("{s: s, s: s}",
+		"event", args.event,
+		"extra", args.extra);
+	if (!blob) {
+		return res;
+	}
+	ast_cel_publish_event(chan, AST_CEL_USER_DEFINED, blob);
 	return res;
 }
 

apps/app_dial.c

@@ -859,8 +859,6 @@ static void do_forward(struct chanlist *o, struct cause_args *num,
 		ast_clear_flag64(o, OPT_IGNORE_CONNECTEDLINE);
 	}
 
-	ast_cel_report_event(in, AST_CEL_FORWARD, NULL, ast_channel_call_forward(c), NULL);
-
 	/* Before processing channel, go ahead and check for forwarding */
 	ast_verb(3, "Now forwarding %s to '%s/%s' (thanks to %s)\n", ast_channel_name(in), tech, stuff, ast_channel_name(c));
 	/* If we have been told to ignore forwards, just set this channel to null and continue processing extensions normally */
@@ -1004,7 +1002,8 @@ static void do_forward(struct chanlist *o, struct cause_args *num,
 			ast_channel_unlock(c);
 
 			ast_channel_lock_both(original, in);
-			ast_channel_publish_dial(in, original, NULL, "CANCEL");
+			ast_channel_publish_dial_forward(in, original, NULL, "CANCEL",
+				ast_channel_call_forward(c));
 			ast_channel_unlock(in);
 			ast_channel_unlock(original);
 

apps/app_queue.c

@@ -4449,8 +4449,6 @@ static struct callattempt *wait_for_answer(struct queue_ent *qe, struct callatte
 						o->block_connected_update = 0;
 					}
 
-					ast_cel_report_event(in, AST_CEL_FORWARD, NULL, ast_channel_call_forward(o->chan), NULL);
-
 					ast_verb(3, "Now forwarding %s to '%s/%s' (thanks to %s)\n", inchan_name, tech, stuff, ochan_name);
 					/* Setup parameters */
 					o->chan = ast_request(tech, ast_channel_nativeformats(in), in, stuff, &status);
@@ -4544,7 +4542,8 @@ static struct callattempt *wait_for_answer(struct queue_ent *qe, struct callatte
 					ast_channel_unlock(qe->chan);
 
 					ast_channel_lock_both(qe->chan, original);
-					ast_channel_publish_dial(qe->chan, original, NULL, "CANCEL");
+					ast_channel_publish_dial_forward(qe->chan, original, NULL, "CANCEL",
+						ast_channel_call_forward(original));
 					ast_channel_unlock(original);
 					ast_channel_unlock(qe->chan);
 					/* Hangup the original channel now, in case we needed it */

include/asterisk/cel.h

@@ -83,8 +83,6 @@ enum ast_cel_event_type {
 	AST_CEL_USER_DEFINED = 21,
 	/*! \brief the last channel with the given linkedid is retired  */
 	AST_CEL_LINKEDID_END = 22,
-	/*! \brief a masquerade happened to alter the participants on a bridge  */
-	AST_CEL_BRIDGE_UPDATE = 23,
 	/*! \brief a directed pickup was performed on this channel  */
 	AST_CEL_PICKUP = 24,
 	/*! \brief this call was forwarded somewhere else  */
@@ -245,6 +243,26 @@ struct ast_cel_event_record {
  */
 int ast_cel_fill_record(const struct ast_event *event, struct ast_cel_event_record *r);
 
+/*!
+ * \brief Publish a CEL event
+ * \since 12
+ *
+ * \param chan This is the primary channel associated with this channel event.
+ * \param event_type This is the type of call event being reported.
+ * \param blob This contains any additional parameters that need to be conveyed for this event.
+ */
+void ast_cel_publish_event(struct ast_channel *chan,
+	enum ast_cel_event_type event_type,
+	struct ast_json *blob);
+
+/*!
+ * \brief Get the CEL topic
+ *
+ * \retval The CEL topic
+ * \retval NULL if not allocated
+ */
+struct stasis_topic *ast_cel_topic(void);
+
 #if defined(__cplusplus) || defined(c_plusplus)
 }
 #endif

include/asterisk/stasis_channels.h

@@ -454,6 +454,24 @@ void ast_channel_publish_dial(struct ast_channel *caller,
 		const char *dialstring,
 		const char *dialstatus);
 
+/*!
+ * \since 12
+ * \brief Publish in the \ref ast_channel_topic or \ref ast_channel_topic_all
+ * topics a stasis message for the channels involved in a dial operation that
+ * is forwarded.
+ *
+ * \param caller The channel performing the dial operation
+ * \param peer The channel being dialed
+ * \param dialstring The information passed to the dialing application when beginning a dial
+ * \param dialstatus The current status of the dial operation
+ * \param forward The call forward string provided by the dialed channel
+ */
+void ast_channel_publish_dial_forward(struct ast_channel *caller,
+		struct ast_channel *peer,
+		const char *dialstring,
+		const char *dialstatus,
+		const char *forward);
+
 /*!
  * \since 12
  * \brief Publish in the \ref ast_channel_topic a \ref ast_channel_snapshot

main/cel.c

@@ -96,7 +96,6 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 						<enum name="APP_END"/>
 						<enum name="BRIDGE_START"/>
 						<enum name="BRIDGE_END"/>
-						<enum name="BRIDGE_UPDATE"/>
 						<enum name="BRIDGE_TO_CONF"/>
 						<enum name="CONF_START"/>
 						<enum name="CONF_END"/>
@@ -126,8 +125,11 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 /*! Message router for state that CEL needs to know about */
 static struct stasis_message_router *cel_state_router;
 
+/*! Topic for CEL-specific messages */
+static struct stasis_topic *cel_topic;
+
 /*! Aggregation topic for all topics CEL needs to know about */
-static struct stasis_topic *cel_state_topic;
+static struct stasis_topic *cel_aggregation_topic;
 
 /*! Subscription for forwarding the channel caching topic */
 static struct stasis_subscription *cel_channel_forwarder;
@@ -138,9 +140,15 @@ static struct stasis_subscription *cel_bridge_forwarder;
 /*! Subscription for forwarding the parking topic */
 static struct stasis_subscription *cel_parking_forwarder;
 
+/*! Subscription for forwarding the CEL-specific topic */
+static struct stasis_subscription *cel_cel_forwarder;
+
 /*! Container for primary channel/bridge ID listing for 2 party bridges */
 static struct ao2_container *bridge_primaries;
 
+struct stasis_message_type *cel_generic_type(void);
+STASIS_MESSAGE_TYPE_DEFN(cel_generic_type);
+
 /*! The number of buckets into which primary channel uniqueids will be hashed */
 #define BRIDGE_PRIMARY_BUCKETS 251
 
@@ -305,7 +313,6 @@ static const char * const cel_event_types[CEL_MAX_EVENT_IDS] = {
 	[AST_CEL_APP_END]          = "APP_END",
 	[AST_CEL_BRIDGE_START]     = "BRIDGE_START",
 	[AST_CEL_BRIDGE_END]       = "BRIDGE_END",
-	[AST_CEL_BRIDGE_UPDATE]    = "BRIDGE_UPDATE",
 	[AST_CEL_BRIDGE_TO_CONF]   = "BRIDGE_TO_CONF",
 	[AST_CEL_CONF_START]       = "CONF_START",
 	[AST_CEL_CONF_END]         = "CONF_END",
@@ -1050,14 +1057,14 @@ static struct ast_multi_channel_blob *get_dialstatus_blob(const char *uniqueid)
 	return ao2_find(cel_dialstatus_store, uniqueid, OBJ_KEY | OBJ_UNLINK);
 }
 
-static const char *get_caller_dialstatus(struct ast_multi_channel_blob *blob)
+static const char *get_blob_variable(struct ast_multi_channel_blob *blob, const char *varname)
 {
 	struct ast_json *json = ast_multi_channel_blob_get_json(blob);
 	if (!json) {
 		return NULL;
 	}
 
-	json = ast_json_object_get(json, "dialstatus");
+	json = ast_json_object_get(json, varname);
 	if (!json) {
 		return NULL;
 	}
@@ -1090,8 +1097,8 @@ static void cel_channel_state_change(
 		RAII_VAR(struct ast_str *, extra_str, ast_str_create(128), ast_free);
 		RAII_VAR(struct ast_multi_channel_blob *, blob, get_dialstatus_blob(new_snapshot->uniqueid), ao2_cleanup);
 		const char *dialstatus = "";
-		if (blob && !ast_strlen_zero(get_caller_dialstatus(blob))) {
-			dialstatus = get_caller_dialstatus(blob);
+		if (blob && !ast_strlen_zero(get_blob_variable(blob, "dialstatus"))) {
+			dialstatus = get_blob_variable(blob, "dialstatus");
 		}
 		ast_str_set(&extra_str, 0, "%d,%s,%s",
 			new_snapshot->hangupcause,
@@ -1406,24 +1413,59 @@ static void cel_dial_cb(void *data, struct stasis_subscription *sub,
 		return;
 	}
 
-	if (ast_strlen_zero(get_caller_dialstatus(blob))) {
+	if (!ast_strlen_zero(get_blob_variable(blob, "forward"))) {
+		struct ast_channel_snapshot *caller = ast_multi_channel_blob_get_channel(blob, "caller");
+		if (!caller) {
+			return;
+		}
+
+		report_event_snapshot(caller, AST_CEL_FORWARD, NULL, get_blob_variable(blob, "forward"), NULL);
+	}
+
+	if (ast_strlen_zero(get_blob_variable(blob, "dialstatus"))) {
 		return;
 	}
 
 	save_dialstatus(blob);
 }
 
+static void cel_generic_cb(
+	void *data, struct stasis_subscription *sub,
+	struct stasis_topic *topic,
+	struct stasis_message *message)
+{
+	struct ast_channel_blob *obj = stasis_message_data(message);
+	int event_type = ast_json_integer_get(ast_json_object_get(obj->blob, "event_type"));
+	struct ast_json *event_details = ast_json_object_get(obj->blob, "event_details");
+
+	switch (event_type) {
+	case AST_CEL_USER_DEFINED:
+		{
+			const char *event = ast_json_string_get(ast_json_object_get(event_details, "event"));
+			const char *extra = ast_json_string_get(ast_json_object_get(event_details, "extra"));
+			report_event_snapshot(obj->snapshot, event_type, event, extra, NULL);
+			break;
+		}
+	default:
+		ast_log(LOG_ERROR, "Unhandled %s event blob\n", ast_cel_get_type_name(event_type));
+		break;
+	}
+}
+
 static void ast_cel_engine_term(void)
 {
 	aco_info_destroy(&cel_cfg_info);
 	ao2_global_obj_release(cel_configs);
 	stasis_message_router_unsubscribe_and_join(cel_state_router);
 	cel_state_router = NULL;
-	ao2_cleanup(cel_state_topic);
-	cel_state_topic = NULL;
+	ao2_cleanup(cel_aggregation_topic);
+	cel_aggregation_topic = NULL;
+	ao2_cleanup(cel_topic);
+	cel_topic = NULL;
 	cel_channel_forwarder = stasis_unsubscribe_and_join(cel_channel_forwarder);
 	cel_bridge_forwarder = stasis_unsubscribe_and_join(cel_bridge_forwarder);
 	cel_parking_forwarder = stasis_unsubscribe_and_join(cel_parking_forwarder);
+	cel_cel_forwarder = stasis_unsubscribe_and_join(cel_cel_forwarder);
 	ao2_cleanup(bridge_primaries);
 	bridge_primaries = NULL;
 	ast_cli_unregister(&cli_status);
@@ -1431,6 +1473,7 @@ static void ast_cel_engine_term(void)
 	cel_dialstatus_store = NULL;
 	ao2_cleanup(linkedids);
 	linkedids = NULL;
+	STASIS_MESSAGE_TYPE_CLEANUP(cel_generic_type);
 }
 
 int ast_cel_engine_init(void)
@@ -1444,6 +1487,10 @@ int ast_cel_engine_init(void)
 		return -1;
 	}
 
+	if (STASIS_MESSAGE_TYPE_INIT(cel_generic_type)) {
+		return -1;
+	}
+
 	if (ast_cli_register(&cli_status)) {
 		return -1;
 	}
@@ -1453,33 +1500,45 @@ int ast_cel_engine_init(void)
 		return -1;
 	}
 
-	cel_state_topic = stasis_topic_create("cel_state_topic");
-	if (!cel_state_topic) {
+	cel_aggregation_topic = stasis_topic_create("cel_aggregation_topic");
+	if (!cel_aggregation_topic) {
+		return -1;
+	}
+
+	cel_topic = stasis_topic_create("cel_topic");
+	if (!cel_topic) {
 		return -1;
 	}
 
 	cel_channel_forwarder = stasis_forward_all(
 		stasis_caching_get_topic(ast_channel_topic_all_cached()),
-		cel_state_topic);
+		cel_aggregation_topic);
 	if (!cel_channel_forwarder) {
 		return -1;
 	}
 
 	cel_bridge_forwarder = stasis_forward_all(
 		stasis_caching_get_topic(ast_bridge_topic_all_cached()),
-		cel_state_topic);
+		cel_aggregation_topic);
 	if (!cel_bridge_forwarder) {
 		return -1;
 	}
 
 	cel_parking_forwarder = stasis_forward_all(
 		ast_parking_topic(),
-		cel_state_topic);
+		cel_aggregation_topic);
 	if (!cel_parking_forwarder) {
 		return -1;
 	}
 
-	cel_state_router = stasis_message_router_create(cel_state_topic);
+	cel_cel_forwarder = stasis_forward_all(
+		ast_cel_topic(),
+		cel_aggregation_topic);
+	if (!cel_cel_forwarder) {
+		return -1;
+	}
+
+	cel_state_router = stasis_message_router_create(cel_aggregation_topic);
 	if (!cel_state_router) {
 		return -1;
 	}
@@ -1509,6 +1568,11 @@ int ast_cel_engine_init(void)
 		cel_parking_cb,
 		NULL);
 
+	ret |= stasis_message_router_add(cel_state_router,
+		cel_generic_type(),
+		cel_generic_cb,
+		NULL);
+
 	/* If somehow we failed to add any routes, just shut down the whole
 	 * thing and fail it.
 	 */
@@ -1538,3 +1602,24 @@ int ast_cel_engine_reload(void)
 	return do_reload();
 }
 
+void ast_cel_publish_event(struct ast_channel *chan,
+	enum ast_cel_event_type event_type,
+	struct ast_json *blob)
+{
+	RAII_VAR(struct ast_channel_blob *, obj, NULL, ao2_cleanup);
+	RAII_VAR(struct ast_json *, cel_blob, NULL, ast_json_unref);
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	cel_blob = ast_json_pack("{s: i, s: O}",
+		"event_type", event_type,
+		"event_details", blob);
+
+	message = ast_channel_blob_create(chan, cel_generic_type(), cel_blob);
+	if (message) {
+		stasis_publish(ast_cel_topic(), message);
+	}
+}
+
+struct stasis_topic *ast_cel_topic(void)
+{
+	return cel_topic;
+}

main/stasis_channels.c

@@ -210,7 +210,8 @@ static void channel_blob_dtor(void *obj)
 	ast_json_unref(event->blob);
 }
 
-void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *peer, const char *dialstring, const char *dialstatus)
+void ast_channel_publish_dial_forward(struct ast_channel *caller, struct ast_channel *peer,
+	const char *dialstring, const char *dialstatus, const char *forward)
 {
 	RAII_VAR(struct ast_multi_channel_blob *, payload, NULL, ao2_cleanup);
 	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
@@ -219,8 +220,9 @@ void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *pe
 	RAII_VAR(struct ast_channel_snapshot *, peer_snapshot, NULL, ao2_cleanup);
 
 	ast_assert(peer != NULL);
-	blob = ast_json_pack("{s: s, s: s}",
+	blob = ast_json_pack("{s: s, s: s, s: s}",
 			     "dialstatus", S_OR(dialstatus, ""),
+			     "forward", S_OR(forward, ""),
 			     "dialstring", S_OR(dialstring, ""));
 	if (!blob) {
 		return;
@@ -252,6 +254,12 @@ void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *pe
 	publish_message_for_channel_topics(msg, caller);
 }
 
+void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *peer,
+	const char *dialstring, const char *dialstatus)
+{
+	ast_channel_publish_dial_forward(caller, peer, dialstring, dialstatus, NULL);
+}
+
 static struct stasis_message *create_channel_blob_message(struct ast_channel_snapshot *snapshot,
 		struct stasis_message_type *type,
 		struct ast_json *blob)