app_cdr,app_forkcdr,func_cdr: Synchronize with engine when manipulating state

When doing the rework of the CDR engine that pushed all of the logic into cdr.c
and made it respond to changes in channel state over Stasis, we knew that
accessing the CDR engine from the dialplan would be "slightly"
non-deterministic. Dialplan threads would be accessing CDRs while Stasis
threads would be updating the state of said CDRs - whereas in the past,
everything happened on the dialplan threads. Tests have shown that "slightly"
is in reality "very".

This patch synchronizes things by making the dialplan applications/functions
that manipulate CDRs do so over Stasis. ForkCDR, NoCDR, ResetCDR, CDR, and
CDR_PROP now all use Stasis to send their requests over to the CDR engine,
and synchronize on the channel Stasis topic via a subscription so that they
return their values/control to the dialplan at the appropriate time.

While going through this, the following changes were also made:
 * DISA, which can reset the CDR when a user successfully authenticates, now
   just uses the ResetCDR app to do this. This prevents having to duplicate
   the same Stasis synchronization logic in that application.
 * Answer no longer disables CDRs. It actually didn't work anyway - calling
   DISABLE on the channel's CDR doesn't stop the CDR from getting the Answer
   time - it just kills all CDRs on that channel, which isn't what the caller
   would intend.

(closes issue ASTERISK-22884)
(closes issue ASTERISK-22886)

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

Merged revisions 404294 from http://svn.asterisk.org/svn/asterisk/branches/12


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

3 files changed, 173 insertions, 20 deletions

diff --git a/apps/app_cdr.c b/apps/app_cdr.c
index 34fd45675..2793846f9 100644
--- a/apps/app_cdr.c
+++ b/apps/app_cdr.c
@@ -36,6 +36,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/channel.h"
 #include "asterisk/module.h"
 #include "asterisk/app.h"
+#include "asterisk/stasis.h"
 
 /*** DOCUMENTATION
 	<application name="NoCDR" language="en_US">
@@ -112,43 +113,130 @@ AST_APP_OPTIONS(resetcdr_opts, {
 	AST_APP_OPTION('e', AST_CDR_FLAG_DISABLE_ALL),
 });
 
+STASIS_MESSAGE_TYPE_DEFN_LOCAL(appcdr_message_type);
+
+/*! \internal \brief Payload for the Stasis message sent to manipulate a CDR */
+struct app_cdr_message_payload {
+	/*! The name of the channel to be manipulated */
+	const char *channel_name;
+	/*! Disable the CDR for this channel */
+	int disable:1;
+	/*! Re-enable the CDR for this channel */
+	int reenable:1;
+	/*! Reset the CDR */
+	int reset:1;
+	/*! If reseting the CDR, keep the variables */
+	int keep_variables:1;
+};
+
+static void appcdr_callback(void *data, struct stasis_subscription *sub, struct stasis_message *message)
+{
+	struct app_cdr_message_payload *payload;
+
+	if (stasis_message_type(message) != appcdr_message_type()) {
+		return;
+	}
+
+	payload = stasis_message_data(message);
+	if (!payload) {
+		return;
+	}
+
+	if (payload->disable) {
+		if (ast_cdr_set_property(payload->channel_name, AST_CDR_FLAG_DISABLE_ALL)) {
+			ast_log(AST_LOG_WARNING, "Failed to disable CDRs on channel %s\n",
+				payload->channel_name);
+		}
+	}
+
+	if (payload->reenable) {
+		if (ast_cdr_clear_property(payload->channel_name, AST_CDR_FLAG_DISABLE_ALL)) {
+			ast_log(AST_LOG_WARNING, "Failed to enable CDRs on channel %s\n",
+				payload->channel_name);
+		}
+	}
+
+	if (payload->reset) {
+		if (ast_cdr_reset(payload->channel_name, payload->keep_variables)) {
+			ast_log(AST_LOG_WARNING, "Failed to reset CDRs on channel %s\n",
+				payload->channel_name);
+		}
+	}
+}
+
+static int publish_app_cdr_message(struct ast_channel *chan, struct app_cdr_message_payload *payload)
+{
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	RAII_VAR(struct stasis_subscription *, subscription, NULL, ao2_cleanup);
+
+	message = stasis_message_create(appcdr_message_type(), payload);
+	if (!message) {
+		ast_log(AST_LOG_WARNING, "Failed to manipulate CDR for channel %s: unable to create message\n",
+			payload->channel_name);
+		return -1;
+	}
+
+	subscription = stasis_subscribe(ast_channel_topic(chan), appcdr_callback, NULL);
+	if (!subscription) {
+		ast_log(AST_LOG_WARNING, "Failed to manipulate CDR for channel %s: unable to create subscription\n",
+			payload->channel_name);
+		return -1;
+	}
+
+	stasis_publish(ast_channel_topic(chan), message);
+
+	subscription = stasis_unsubscribe_and_join(subscription);
+	return 0;
+}
+
 static int resetcdr_exec(struct ast_channel *chan, const char *data)
 {
+	RAII_VAR(struct app_cdr_message_payload *, payload,
+		ao2_alloc(sizeof(*payload), NULL), ao2_cleanup);
 	char *args;
 	struct ast_flags flags = { 0 };
-	int res = 0;
+
+	if (!payload) {
+		return -1;
+	}
 
 	if (!ast_strlen_zero(data)) {
 		args = ast_strdupa(data);
 		ast_app_parse_options(resetcdr_opts, &flags, NULL, args);
 	}
 
+	payload->channel_name = ast_channel_name(chan);
+	payload->reset = 1;
+
 	if (ast_test_flag(&flags, AST_CDR_FLAG_DISABLE_ALL)) {
-		if (ast_cdr_clear_property(ast_channel_name(chan), AST_CDR_FLAG_DISABLE_ALL)) {
-			res = 1;
-		}
-	}
-	if (ast_cdr_reset(ast_channel_name(chan), &flags)) {
-		res = 1;
+		payload->reenable = 1;
 	}
 
-	if (res) {
-		ast_log(AST_LOG_WARNING, "Failed to reset CDR for channel %s\n", ast_channel_name(chan));
+	if (ast_test_flag(&flags, AST_CDR_FLAG_KEEP_VARS)) {
+		payload->keep_variables = 1;
 	}
-	return res;
+
+	return publish_app_cdr_message(chan, payload);
 }
 
 static int nocdr_exec(struct ast_channel *chan, const char *data)
 {
-	if (ast_cdr_set_property(ast_channel_name(chan), AST_CDR_FLAG_DISABLE_ALL)) {
-		ast_log(AST_LOG_WARNING, "Failed to disable CDR for channel %s\n", ast_channel_name(chan));
+	RAII_VAR(struct app_cdr_message_payload *, payload,
+		ao2_alloc(sizeof(*payload), NULL), ao2_cleanup);
+
+	if (!payload) {
+		return -1;
 	}
 
-	return 0;
+	payload->channel_name = ast_channel_name(chan);
+	payload->disable = 1;
+
+	return publish_app_cdr_message(chan, payload);
 }
 
 static int unload_module(void)
 {
+	STASIS_MESSAGE_TYPE_CLEANUP(appcdr_message_type);
 	ast_unregister_application(nocdr_app);
 	ast_unregister_application(resetcdr_app);
 	return 0;
@@ -158,6 +246,7 @@ static int load_module(void)
 {
 	int res = 0;
 
+	res |= STASIS_MESSAGE_TYPE_INIT(appcdr_message_type);
 	res |= ast_register_application_xml(nocdr_app, nocdr_exec);
 	res |= ast_register_application_xml(resetcdr_app, resetcdr_exec);
 
diff --git a/apps/app_disa.c b/apps/app_disa.c
index 9e7412717..824e8fe55 100644
--- a/apps/app_disa.c
+++ b/apps/app_disa.c
@@ -27,6 +27,7 @@
  */
 
 /*** MODULEINFO
+	<use type="module">app_cdr</use>
 	<support_level>core</support_level>
  ***/
 
@@ -362,7 +363,7 @@ static int disa_exec(struct ast_channel *chan, const char *data)
 
 	if (k == 3) {
 		int recheck = 0;
-		struct ast_flags cdr_flags = { AST_CDR_FLAG_DISABLE, };
+		struct ast_app *app_reset_cdr;
 
 		if (!ast_exists_extension(chan, args.context, exten, 1,
 			S_COR(ast_channel_caller(chan)->id.number.valid, ast_channel_caller(chan)->id.number.str, NULL))) {
@@ -387,10 +388,12 @@ static int disa_exec(struct ast_channel *chan, const char *data)
 				ast_channel_unlock(chan);
 			}
 
-			if (special_noanswer) {
-				ast_clear_flag(&cdr_flags, AST_CDR_FLAG_DISABLE);
+			app_reset_cdr = pbx_findapp("ResetCDR");
+			if (app_reset_cdr) {
+				pbx_exec(chan, app_reset_cdr, special_noanswer ? "" : "e");
+			} else {
+				ast_log(AST_LOG_NOTICE, "ResetCDR application not found; CDR will not be reset\n");
 			}
-			ast_cdr_reset(ast_channel_name(chan), &cdr_flags);
 			ast_explicit_goto(chan, args.context, exten, 1);
 			return 0;
 		}
diff --git a/apps/app_forkcdr.c b/apps/app_forkcdr.c
index 6231d381f..af5ae6a1c 100644
--- a/apps/app_forkcdr.c
+++ b/apps/app_forkcdr.c
@@ -40,6 +40,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/cdr.h"
 #include "asterisk/app.h"
 #include "asterisk/module.h"
+#include "asterisk/stasis.h"
 
 /*** DOCUMENTATION
 	<application name="ForkCDR" language="en_US">
@@ -102,8 +103,41 @@ AST_APP_OPTIONS(forkcdr_exec_options, {
 	AST_APP_OPTION('v', AST_CDR_FLAG_KEEP_VARS),
 });
 
+STASIS_MESSAGE_TYPE_DEFN_LOCAL(forkcdr_message_type);
+
+/*! \internal \brief Message payload for the Stasis message sent to fork the CDR */
+struct fork_cdr_message_payload {
+	/*! The name of the channel whose CDR will be forked */
+	const char *channel_name;
+	/*! Option flags that control how the CDR will be forked */
+	struct ast_flags *flags;
+};
+
+static void forkcdr_callback(void *data, struct stasis_subscription *sub, struct stasis_message *message)
+{
+	struct fork_cdr_message_payload *payload;
+
+	if (stasis_message_type(message) != forkcdr_message_type()) {
+		return;
+	}
+
+	payload = stasis_message_data(message);
+	if (!payload) {
+		return;
+	}
+
+	if (ast_cdr_fork(payload->channel_name, payload->flags)) {
+		ast_log(AST_LOG_WARNING, "Failed to fork CDR for channel %s\n",
+			payload->channel_name);
+	}
+}
+
 static int forkcdr_exec(struct ast_channel *chan, const char *data)
 {
+	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	RAII_VAR(struct fork_cdr_message_payload *, payload, ao2_alloc(sizeof(*payload), NULL), ao2_cleanup);
+	RAII_VAR(struct stasis_subscription *, subscription, NULL, ao2_cleanup);
+
 	char *parse;
 	struct ast_flags flags = { 0, };
 	AST_DECLARE_APP_ARGS(args,
@@ -118,21 +152,48 @@ static int forkcdr_exec(struct ast_channel *chan, const char *data)
 		ast_app_parse_options(forkcdr_exec_options, &flags, NULL, args.options);
 	}
 
-	if (ast_cdr_fork(ast_channel_name(chan), &flags)) {
-		ast_log(AST_LOG_WARNING, "Failed to fork CDR for channel %s\n", ast_channel_name(chan));
+	if (!payload) {
+		return -1;
+	}
+
+	payload->channel_name = ast_channel_name(chan);
+	payload->flags = &flags;
+	message = stasis_message_create(forkcdr_message_type(), payload);
+	if (!message) {
+		ast_log(AST_LOG_WARNING, "Failed to fork CDR for channel %s: unable to create message\n",
+			ast_channel_name(chan));
+		return -1;
+	}
+
+	subscription = stasis_subscribe(ast_channel_topic(chan), forkcdr_callback, NULL);
+	if (!subscription) {
+		ast_log(AST_LOG_WARNING, "Failed to fork CDR for channel %s: unable to create subscription\n",
+			payload->channel_name);
+		return -1;
 	}
 
+	stasis_publish(ast_channel_topic(chan), message);
+
+	subscription = stasis_unsubscribe_and_join(subscription);
+
 	return 0;
 }
 
 static int unload_module(void)
 {
+	STASIS_MESSAGE_TYPE_CLEANUP(forkcdr_message_type);
+
 	return ast_unregister_application(app);
 }
 
 static int load_module(void)
 {
-	return ast_register_application_xml(app, forkcdr_exec);
+	int res = 0;
+
+	res |= STASIS_MESSAGE_TYPE_INIT(forkcdr_message_type);
+	res |= ast_register_application_xml(app, forkcdr_exec);
+
+	return res;
 }
 
 AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "Fork The CDR into 2 separate entities");