kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-09-04 20:04:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

CDRs: Synchronize dialplan applications that manipulate CDRs with the engine

Browse Source 
In https://reviewboard.asterisk.org/r/3057/, applications and functions that
manipulate CDRs were made to interact over Stasis. This was done to
synchronize manipulations of CDRs from the dialplan with the updates the
engine itself receives over the message bus.

This change rested on a faulty premise: that messages published to the CDR
topic or to a topic that forwards to the CDR topic are synchronized with the
messages handled by the CDR topic subscription in the CDR engine. This is not
the case. There is no ordering guaranteed for two messages published to the
same topic; ordering is only guaranteed if a message is published to the same
subscriber.

Stasis was modified in r405311 to allow a publisher to synchronize on the
subscriber. This patch uses that API to synchronize the CDR publishers with
the CDR engine message router, which maintains the overall topic subscription.

(closes issue ASTERISK-22884)
Reported by: Matt Jordan

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

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


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@405314 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
Matthew Jordan
2014-01-12 22:13:12 +00:00
parent f8aaf585a3
commit 373965dbff
5 changed files with 148 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
apps/app_cdr.c
View File

@@ -37,6 +37,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/module.h"
#include "asterisk/app.h"
#include "asterisk/stasis.h"
#include "asterisk/stasis_message_router.h"
/*** DOCUMENTATION
<application name="NoCDR" language="en_US">
@@ -167,7 +168,13 @@ static void appcdr_callback(void *data, struct stasis_subscription *sub, struct
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);
RAII_VAR(struct stasis_message_router *, router, ast_cdr_message_router(), ao2_cleanup);
if (!router) {
ast_log(AST_LOG_WARNING, "Failed to manipulate CDR for channel %s: no message router\n",
ast_channel_name(chan));
return -1;
}
message = stasis_message_create(appcdr_message_type(), payload);
if (!message) {
@@ -175,17 +182,8 @@ static int publish_app_cdr_message(struct ast_channel *chan, struct app_cdr_mess
payload->channel_name);
return -1;
}
stasis_message_router_publish_sync(router, message);
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;
}
@@ -236,6 +234,11 @@ static int nocdr_exec(struct ast_channel *chan, const char *data)
static int unload_module(void)
{
RAII_VAR(struct stasis_message_router *, router, ast_cdr_message_router(), ao2_cleanup);
if (router) {
stasis_message_router_remove(router, appcdr_message_type());
}
STASIS_MESSAGE_TYPE_CLEANUP(appcdr_message_type);
ast_unregister_application(nocdr_app);
ast_unregister_application(resetcdr_app);
@@ -244,11 +247,18 @@ static int unload_module(void)
static int load_module(void)
{
RAII_VAR(struct stasis_message_router *, router, ast_cdr_message_router(), ao2_cleanup);
int res = 0;
if (!router) {
return AST_MODULE_LOAD_FAILURE;
}
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);
res |= stasis_message_router_add(router, appcdr_message_type(),
appcdr_callback, NULL);
if (res) {
return AST_MODULE_LOAD_FAILURE;

42
apps/app_forkcdr.c
View File

@@ -41,6 +41,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/app.h"
#include "asterisk/module.h"
#include "asterisk/stasis.h"
#include "asterisk/stasis_message_router.h"
/*** DOCUMENTATION
<application name="ForkCDR" language="en_US">
@@ -136,7 +137,7 @@ 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);
RAII_VAR(struct stasis_message_router *, router, ast_cdr_message_router(), ao2_cleanup);
char *parse;
struct ast_flags flags = { 0, };
@@ -156,6 +157,12 @@ static int forkcdr_exec(struct ast_channel *chan, const char *data)
return -1;
}
if (!router) {
ast_log(AST_LOG_WARNING, "Failed to manipulate CDR for channel %s: no message router\n",
ast_channel_name(chan));
return -1;
}
payload->channel_name = ast_channel_name(chan);
payload->flags = &flags;
message = stasis_message_create(forkcdr_message_type(), payload);
@@ -164,36 +171,41 @@ static int forkcdr_exec(struct ast_channel *chan, const char *data)
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);
stasis_message_router_publish_sync(router, message);
return 0;
}
static int unload_module(void)
{
STASIS_MESSAGE_TYPE_CLEANUP(forkcdr_message_type);
RAII_VAR(struct stasis_message_router *, router, ast_cdr_message_router(), ao2_cleanup);
return ast_unregister_application(app);
if (router) {
stasis_message_router_remove(router, forkcdr_message_type());
}
STASIS_MESSAGE_TYPE_CLEANUP(forkcdr_message_type);
ast_unregister_application(app);
return 0;
}
static int load_module(void)
{
RAII_VAR(struct stasis_message_router *, router, ast_cdr_message_router(), ao2_cleanup);
int res = 0;
if (!router) {
return AST_MODULE_LOAD_FAILURE;
}
res |= STASIS_MESSAGE_TYPE_INIT(forkcdr_message_type);
res |= ast_register_application_xml(app, forkcdr_exec);
res |= stasis_message_router_add(router, forkcdr_message_type(),
forkcdr_callback, NULL);
return res;
if (res) {
return AST_MODULE_LOAD_FAILURE;
}
return AST_MODULE_LOAD_SUCCESS;
}
AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "Fork The CDR into 2 separate entities");