Asterisk
/
asterisk
mirror of https://github.com/asterisk/asterisk
1
0
Fork 0
Code Issues Releases Wiki Activity

Make app_queue and res_agi independent of AMI being enabled.

Browse Source 
The https://reviewboard.asterisk.org/r/2888/ review changes manager to not
subscribe to stasis when it is disabled for performance reasons.  When
manager is disabled app_queue and res_agi decline to load and fail to
clean up what they have already allocated.

* Made app_queue and res_agi clean up allocated resources when they
decline to load.

* Made app_queue and res_agi use their own subscriptions to the stasis
topics instead of borrowing manager's message router structure
inappropriately.

(closes issue ASTERISK-22604)
Reported by: rmudgett

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

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


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@400672 65c4cc65-6c06-0410-ace0-fbb531ad65f3
changes/97/197/1
Richard Mudgett 12 years ago
parent 96d27333d2
commit 8eec8fbf83
2 changed files with 321 additions and 335 deletions
Show Stats Download Patch File Download Diff File

486
apps/app_queue.c
Unescape View File

@ -1818,54 +1818,116 @@ static inline void insert_entry(struct call_queue *q, struct queue_ent *prev, st
new->opos = *pos; new->opos = *pos;
} }
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_caller_join_type); static struct ast_manager_event_blob *queue_channel_to_ami(const char *type, struct stasis_message *message)
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_caller_leave_type); {
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_caller_abandon_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_status_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_added_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_removed_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_pause_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_penalty_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_ringinuse_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_called_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_connect_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_complete_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_dump_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_ringnoanswer_type);
static void queue_channel_manager_event(void *data,
struct stasis_subscription *sub,
struct stasis_message *message)
{
const char *type = data;
struct ast_channel_blob *obj = stasis_message_data(message); struct ast_channel_blob *obj = stasis_message_data(message);
RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, channel_string, NULL, ast_free);
RAII_VAR(struct ast_str *, event_string, NULL, ast_free); RAII_VAR(struct ast_str *, event_string, NULL, ast_free);
channel_event_string = ast_manager_build_channel_state_string(obj->snapshot); channel_string = ast_manager_build_channel_state_string(obj->snapshot);
if (!channel_event_string) { event_string = ast_manager_str_from_json_object(obj->blob, NULL);
return; if (!channel_string || !event_string) {
return NULL;
} }
event_string = ast_manager_str_from_json_object(obj->blob, NULL); return ast_manager_event_blob_create(EVENT_FLAG_AGENT, type,
"%s"
"%s",
ast_str_buffer(channel_string),
ast_str_buffer(event_string));
}
static struct ast_manager_event_blob *queue_caller_join_to_ami(struct stasis_message *message)
{
return queue_channel_to_ami("QueueCallerJoin", message);
}
static struct ast_manager_event_blob *queue_caller_leave_to_ami(struct stasis_message *message)
{
return queue_channel_to_ami("QueueCallerLeave", message);
}
static struct ast_manager_event_blob *queue_caller_abandon_to_ami(struct stasis_message *message)
{
return queue_channel_to_ami("QueueCallerAbandon", message);
}
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_caller_join_type,
.to_ami = queue_caller_join_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_caller_leave_type,
.to_ami = queue_caller_leave_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_caller_abandon_type,
.to_ami = queue_caller_abandon_to_ami,
);
static struct ast_manager_event_blob *queue_member_to_ami(const char *type, struct stasis_message *message)
{
struct ast_json_payload *payload = stasis_message_data(message);
RAII_VAR(struct ast_str *, event_string, NULL, ast_free);
event_string = ast_manager_str_from_json_object(payload->json, NULL);
if (!event_string) { if (!event_string) {
return; return NULL;
} }
manager_event(EVENT_FLAG_AGENT, type, return ast_manager_event_blob_create(EVENT_FLAG_AGENT, type,
"%s"
"%s", "%s",
ast_str_buffer(channel_event_string),
ast_str_buffer(event_string)); ast_str_buffer(event_string));
} }
static void queue_multi_channel_manager_event(void *data, static struct ast_manager_event_blob *queue_member_status_to_ami(struct stasis_message *message)
struct stasis_subscription *sub, {
struct stasis_message *message) return queue_member_to_ami("QueueMemberStatus", message);
}
static struct ast_manager_event_blob *queue_member_added_to_ami(struct stasis_message *message)
{
return queue_member_to_ami("QueueMemberAdded", message);
}
static struct ast_manager_event_blob *queue_member_removed_to_ami(struct stasis_message *message)
{
return queue_member_to_ami("QueueMemberRemoved", message);
}
static struct ast_manager_event_blob *queue_member_pause_to_ami(struct stasis_message *message)
{
return queue_member_to_ami("QueueMemberPause", message);
}
static struct ast_manager_event_blob *queue_member_penalty_to_ami(struct stasis_message *message)
{
return queue_member_to_ami("QueueMemberPenalty", message);
}
static struct ast_manager_event_blob *queue_member_ringinuse_to_ami(struct stasis_message *message)
{
return queue_member_to_ami("QueueMemberRinginuse", message);
}
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_status_type,
.to_ami = queue_member_status_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_added_type,
.to_ami = queue_member_added_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_removed_type,
.to_ami = queue_member_removed_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_pause_type,
.to_ami = queue_member_pause_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_penalty_type,
.to_ami = queue_member_penalty_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_member_ringinuse_type,
.to_ami = queue_member_ringinuse_to_ami,
);
static struct ast_manager_event_blob *queue_multi_channel_to_ami(const char *type, struct stasis_message *message)
{ {
const char *type = data;
struct ast_multi_channel_blob *obj = stasis_message_data(message); struct ast_multi_channel_blob *obj = stasis_message_data(message);
struct ast_channel_snapshot *caller; struct ast_channel_snapshot *caller;
struct ast_channel_snapshot *agent; struct ast_channel_snapshot *agent;
@ -1877,8 +1939,8 @@ static void queue_multi_channel_manager_event(void *data,
if (caller) { if (caller) {
caller_event_string = ast_manager_build_channel_state_string(caller); caller_event_string = ast_manager_build_channel_state_string(caller);
if (!caller_event_string) { if (!caller_event_string) {
ast_log(AST_LOG_NOTICE, "No caller event string, bailing\n"); ast_log(LOG_NOTICE, "No caller event string, bailing\n");
return; return NULL;
} }
} }
@ -1886,17 +1948,17 @@ static void queue_multi_channel_manager_event(void *data,
if (agent) { if (agent) {
agent_event_string = ast_manager_build_channel_state_string_prefix(agent, "Dest"); agent_event_string = ast_manager_build_channel_state_string_prefix(agent, "Dest");
if (!agent_event_string) { if (!agent_event_string) {
ast_log(AST_LOG_NOTICE, "No agent event string, bailing\n"); ast_log(LOG_NOTICE, "No agent event string, bailing\n");
return; return NULL;
} }
} }
event_string = ast_manager_str_from_json_object(ast_multi_channel_blob_get_json(obj), NULL); event_string = ast_manager_str_from_json_object(ast_multi_channel_blob_get_json(obj), NULL);
if (!event_string) { if (!event_string) {
return; return NULL;
} }
manager_event(EVENT_FLAG_AGENT, type, return ast_manager_event_blob_create(EVENT_FLAG_AGENT, type,
"%s" "%s"
"%s" "%s"
"%s", "%s",
@ -1905,24 +1967,47 @@ static void queue_multi_channel_manager_event(void *data,
ast_str_buffer(event_string)); ast_str_buffer(event_string));
} }
static void queue_member_manager_event(void *data, static struct ast_manager_event_blob *queue_agent_called_to_ami(struct stasis_message *message)
struct stasis_subscription *sub,
struct stasis_message *message)
{ {
const char *type = data; return queue_multi_channel_to_ami("AgentCalled", message);
struct ast_json_payload *payload = stasis_message_data(message); }
struct ast_json *event = payload->json;
RAII_VAR(struct ast_str *, event_string, NULL, ast_free);
event_string = ast_manager_str_from_json_object(event, NULL); static struct ast_manager_event_blob *queue_agent_connect_to_ami(struct stasis_message *message)
if (!event_string) { {
return; return queue_multi_channel_to_ami("AgentConnect", message);
}
static struct ast_manager_event_blob *queue_agent_complete_to_ami(struct stasis_message *message)
{
return queue_multi_channel_to_ami("AgentComplete", message);
} }
manager_event(EVENT_FLAG_AGENT, type, static struct ast_manager_event_blob *queue_agent_dump_to_ami(struct stasis_message *message)
"%s", ast_str_buffer(event_string)); {
return queue_multi_channel_to_ami("AgentDump", message);
}
static struct ast_manager_event_blob *queue_agent_ringnoanswer_to_ami(struct stasis_message *message)
{
return queue_multi_channel_to_ami("AgentRingNoAnswer", message);
} }
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_called_type,
.to_ami = queue_agent_called_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_connect_type,
.to_ami = queue_agent_connect_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_complete_type,
.to_ami = queue_agent_complete_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_dump_type,
.to_ami = queue_agent_dump_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(queue_agent_ringnoanswer_type,
.to_ami = queue_agent_ringnoanswer_to_ami,
);
static void queue_publish_multi_channel_snapshot_blob(struct stasis_topic *topic, static void queue_publish_multi_channel_snapshot_blob(struct stasis_topic *topic,
struct ast_channel_snapshot *caller_snapshot, struct ast_channel_snapshot *caller_snapshot,
struct ast_channel_snapshot *agent_snapshot, struct ast_channel_snapshot *agent_snapshot,
@ -10344,30 +10429,9 @@ static struct stasis_forward *topic_forwarder;
static int unload_module(void) static int unload_module(void)
{ {
int res;
struct ao2_iterator q_iter;
struct call_queue *q = NULL;
struct stasis_message_router *message_router;
message_router = ast_manager_get_message_router();
if (message_router) {
stasis_message_router_remove(message_router, queue_caller_join_type());
stasis_message_router_remove(message_router, queue_caller_leave_type());
stasis_message_router_remove(message_router, queue_caller_abandon_type());
stasis_message_router_remove(message_router, queue_member_status_type());
stasis_message_router_remove(message_router, queue_member_added_type());
stasis_message_router_remove(message_router, queue_member_removed_type());
stasis_message_router_remove(message_router, queue_member_pause_type());
stasis_message_router_remove(message_router, queue_member_penalty_type());
stasis_message_router_remove(message_router, queue_member_ringinuse_type());
stasis_message_router_remove(message_router, queue_agent_called_type());
stasis_message_router_remove(message_router, queue_agent_connect_type());
stasis_message_router_remove(message_router, queue_agent_complete_type());
stasis_message_router_remove(message_router, queue_agent_dump_type());
stasis_message_router_remove(message_router, queue_agent_ringnoanswer_type());
}
stasis_message_router_unsubscribe_and_join(agent_router); stasis_message_router_unsubscribe_and_join(agent_router);
agent_router = NULL;
topic_forwarder = stasis_forward_cancel(topic_forwarder); topic_forwarder = stasis_forward_cancel(topic_forwarder);
STASIS_MESSAGE_TYPE_CLEANUP(queue_caller_join_type); STASIS_MESSAGE_TYPE_CLEANUP(queue_caller_join_type);
@ -10388,47 +10452,42 @@ static int unload_module(void)
STASIS_MESSAGE_TYPE_CLEANUP(queue_agent_ringnoanswer_type); STASIS_MESSAGE_TYPE_CLEANUP(queue_agent_ringnoanswer_type);
ast_cli_unregister_multiple(cli_queue, ARRAY_LEN(cli_queue)); ast_cli_unregister_multiple(cli_queue, ARRAY_LEN(cli_queue));
res = ast_manager_unregister("QueueStatus"); ast_manager_unregister("QueueStatus");
res |= ast_manager_unregister("Queues"); ast_manager_unregister("Queues");
res |= ast_manager_unregister("QueueRule"); ast_manager_unregister("QueueRule");
res |= ast_manager_unregister("QueueSummary"); ast_manager_unregister("QueueSummary");
res |= ast_manager_unregister("QueueAdd"); ast_manager_unregister("QueueAdd");
res |= ast_manager_unregister("QueueRemove"); ast_manager_unregister("QueueRemove");
res |= ast_manager_unregister("QueuePause"); ast_manager_unregister("QueuePause");
res |= ast_manager_unregister("QueueLog"); ast_manager_unregister("QueueLog");
res |= ast_manager_unregister("QueuePenalty"); ast_manager_unregister("QueuePenalty");
res |= ast_manager_unregister("QueueReload"); ast_manager_unregister("QueueReload");
res |= ast_manager_unregister("QueueReset"); ast_manager_unregister("QueueReset");
res |= ast_manager_unregister("QueueMemberRingInUse"); ast_manager_unregister("QueueMemberRingInUse");
res |= ast_unregister_application(app_aqm); ast_unregister_application(app_aqm);
res |= ast_unregister_application(app_rqm); ast_unregister_application(app_rqm);
res |= ast_unregister_application(app_pqm); ast_unregister_application(app_pqm);
res |= ast_unregister_application(app_upqm); ast_unregister_application(app_upqm);
res |= ast_unregister_application(app_ql); ast_unregister_application(app_ql);
res |= ast_unregister_application(app); ast_unregister_application(app);
res |= ast_custom_function_unregister(&queueexists_function); ast_custom_function_unregister(&queueexists_function);
res |= ast_custom_function_unregister(&queuevar_function); ast_custom_function_unregister(&queuevar_function);
res |= ast_custom_function_unregister(&queuemembercount_function); ast_custom_function_unregister(&queuemembercount_function);
res |= ast_custom_function_unregister(&queuemembercount_dep); ast_custom_function_unregister(&queuemembercount_dep);
res |= ast_custom_function_unregister(&queuememberlist_function); ast_custom_function_unregister(&queuememberlist_function);
res |= ast_custom_function_unregister(&queuewaitingcount_function); ast_custom_function_unregister(&queuewaitingcount_function);
res |= ast_custom_function_unregister(&queuememberpenalty_function); ast_custom_function_unregister(&queuememberpenalty_function);
res |= ast_data_unregister(NULL); ast_data_unregister(NULL);
device_state_sub = stasis_unsubscribe_and_join(device_state_sub); device_state_sub = stasis_unsubscribe_and_join(device_state_sub);
ast_extension_state_del(0, extension_state_cb); ast_extension_state_del(0, extension_state_cb);
q_iter = ao2_iterator_init(queues, 0);
while ((q = ao2_t_iterator_next(&q_iter, "Iterate through queues"))) {
queues_t_unlink(queues, q, "Remove queue from container due to unload");
queue_t_unref(q, "Done with iterator");
}
ao2_iterator_destroy(&q_iter);
ao2_ref(queues, -1);
ast_unload_realtime("queue_members"); ast_unload_realtime("queue_members");
return res; ao2_cleanup(queues);
queues = NULL;
return 0;
} }
/*! /*!
@ -10443,19 +10502,23 @@ static int unload_module(void)
*/ */
static int load_module(void) static int load_module(void)
{ {
int res; int err = 0;
struct ast_flags mask = {AST_FLAGS_ALL, }; struct ast_flags mask = {AST_FLAGS_ALL, };
struct ast_config *member_config; struct ast_config *member_config;
struct stasis_message_router *manager_router;
struct stasis_topic *queue_topic; struct stasis_topic *queue_topic;
struct stasis_topic *manager_topic; struct stasis_topic *manager_topic;
queues = ao2_container_alloc(MAX_QUEUE_BUCKETS, queue_hash_cb, queue_cmp_cb); queues = ao2_container_alloc(MAX_QUEUE_BUCKETS, queue_hash_cb, queue_cmp_cb);
if (!queues) {
return AST_MODULE_LOAD_DECLINE;
}
use_weight = 0; use_weight = 0;
if (reload_handler(0, &mask, NULL)) if (reload_handler(0, &mask, NULL)) {
unload_module();
return AST_MODULE_LOAD_DECLINE; return AST_MODULE_LOAD_DECLINE;
}
ast_realtime_require_field("queue_members", "paused", RQ_INTEGER1, 1, "uniqueid", RQ_UINTEGER2, 5, SENTINEL); ast_realtime_require_field("queue_members", "paused", RQ_INTEGER1, 1, "uniqueid", RQ_UINTEGER2, 5, SENTINEL);
@ -10468,6 +10531,7 @@ static int load_module(void)
realtime_ringinuse_field = "ringinuse"; realtime_ringinuse_field = "ringinuse";
} else { } else {
const char *config_val; const char *config_val;
if ((config_val = ast_variable_retrieve(member_config, NULL, "ringinuse"))) { if ((config_val = ast_variable_retrieve(member_config, NULL, "ringinuse"))) {
ast_log(LOG_NOTICE, "ringinuse field entries found in queue_members table. Using 'ringinuse'\n"); ast_log(LOG_NOTICE, "ringinuse field entries found in queue_members table. Using 'ringinuse'\n");
realtime_ringinuse_field = "ringinuse"; realtime_ringinuse_field = "ringinuse";
@ -10479,168 +10543,102 @@ static int load_module(void)
realtime_ringinuse_field = "ringinuse"; realtime_ringinuse_field = "ringinuse";
} }
} }
ast_config_destroy(member_config); ast_config_destroy(member_config);
if (queue_persistent_members) if (queue_persistent_members) {
reload_queue_members(); reload_queue_members();
}
ast_data_register_multiple(queue_data_providers, ARRAY_LEN(queue_data_providers)); ast_data_register_multiple(queue_data_providers, ARRAY_LEN(queue_data_providers));
ast_cli_register_multiple(cli_queue, ARRAY_LEN(cli_queue)); err |= ast_cli_register_multiple(cli_queue, ARRAY_LEN(cli_queue));
res = ast_register_application_xml(app, queue_exec); err |= ast_register_application_xml(app, queue_exec);
res |= ast_register_application_xml(app_aqm, aqm_exec); err |= ast_register_application_xml(app_aqm, aqm_exec);
res |= ast_register_application_xml(app_rqm, rqm_exec); err |= ast_register_application_xml(app_rqm, rqm_exec);
res |= ast_register_application_xml(app_pqm, pqm_exec); err |= ast_register_application_xml(app_pqm, pqm_exec);
res |= ast_register_application_xml(app_upqm, upqm_exec); err |= ast_register_application_xml(app_upqm, upqm_exec);
res |= ast_register_application_xml(app_ql, ql_exec); err |= ast_register_application_xml(app_ql, ql_exec);
res |= ast_manager_register_xml("Queues", 0, manager_queues_show); err |= ast_manager_register_xml("Queues", 0, manager_queues_show);
res |= ast_manager_register_xml("QueueStatus", 0, manager_queues_status); err |= ast_manager_register_xml("QueueStatus", 0, manager_queues_status);
res |= ast_manager_register_xml("QueueSummary", 0, manager_queues_summary); err |= ast_manager_register_xml("QueueSummary", 0, manager_queues_summary);
res |= ast_manager_register_xml("QueueAdd", EVENT_FLAG_AGENT, manager_add_queue_member); err |= ast_manager_register_xml("QueueAdd", EVENT_FLAG_AGENT, manager_add_queue_member);
res |= ast_manager_register_xml("QueueRemove", EVENT_FLAG_AGENT, manager_remove_queue_member); err |= ast_manager_register_xml("QueueRemove", EVENT_FLAG_AGENT, manager_remove_queue_member);
res |= ast_manager_register_xml("QueuePause", EVENT_FLAG_AGENT, manager_pause_queue_member); err |= ast_manager_register_xml("QueuePause", EVENT_FLAG_AGENT, manager_pause_queue_member);
res |= ast_manager_register_xml("QueueLog", EVENT_FLAG_AGENT, manager_queue_log_custom); err |= ast_manager_register_xml("QueueLog", EVENT_FLAG_AGENT, manager_queue_log_custom);
res |= ast_manager_register_xml("QueuePenalty", EVENT_FLAG_AGENT, manager_queue_member_penalty); err |= ast_manager_register_xml("QueuePenalty", EVENT_FLAG_AGENT, manager_queue_member_penalty);
res |= ast_manager_register_xml("QueueMemberRingInUse", EVENT_FLAG_AGENT, manager_queue_member_ringinuse); err |= ast_manager_register_xml("QueueMemberRingInUse", EVENT_FLAG_AGENT, manager_queue_member_ringinuse);
res |= ast_manager_register_xml("QueueRule", 0, manager_queue_rule_show); err |= ast_manager_register_xml("QueueRule", 0, manager_queue_rule_show);
res |= ast_manager_register_xml("QueueReload", 0, manager_queue_reload); err |= ast_manager_register_xml("QueueReload", 0, manager_queue_reload);
res |= ast_manager_register_xml("QueueReset", 0, manager_queue_reset); err |= ast_manager_register_xml("QueueReset", 0, manager_queue_reset);
res |= ast_custom_function_register(&queuevar_function); err |= ast_custom_function_register(&queuevar_function);
res |= ast_custom_function_register(&queueexists_function); err |= ast_custom_function_register(&queueexists_function);
res |= ast_custom_function_register(&queuemembercount_function); err |= ast_custom_function_register(&queuemembercount_function);
res |= ast_custom_function_register(&queuemembercount_dep); err |= ast_custom_function_register(&queuemembercount_dep);
res |= ast_custom_function_register(&queuememberlist_function); err |= ast_custom_function_register(&queuememberlist_function);
res |= ast_custom_function_register(&queuewaitingcount_function); err |= ast_custom_function_register(&queuewaitingcount_function);
res |= ast_custom_function_register(&queuememberpenalty_function); err |= ast_custom_function_register(&queuememberpenalty_function);
/* in the following subscribe call, do I use DEVICE_STATE, or DEVICE_STATE_CHANGE? */ /* in the following subscribe call, do I use DEVICE_STATE, or DEVICE_STATE_CHANGE? */
if (!(device_state_sub = stasis_subscribe(ast_device_state_topic_all(), device_state_cb, NULL))) { device_state_sub = stasis_subscribe(ast_device_state_topic_all(), device_state_cb, NULL);
res = -1; if (!device_state_sub) {
err = -1;
} }
manager_topic = ast_manager_get_topic(); manager_topic = ast_manager_get_topic();
if (!manager_topic) {
return AST_MODULE_LOAD_DECLINE;
}
manager_router = ast_manager_get_message_router();
if (!manager_router) {
return AST_MODULE_LOAD_DECLINE;
}
queue_topic = ast_queue_topic_all(); queue_topic = ast_queue_topic_all();
if (!queue_topic) { if (!manager_topic || !queue_topic) {
unload_module();
return AST_MODULE_LOAD_DECLINE; return AST_MODULE_LOAD_DECLINE;
} }
topic_forwarder = stasis_forward_all(queue_topic, manager_topic); topic_forwarder = stasis_forward_all(queue_topic, manager_topic);
if (!topic_forwarder) { if (!topic_forwarder) {
unload_module();
return AST_MODULE_LOAD_DECLINE; return AST_MODULE_LOAD_DECLINE;
} }
if (!ast_channel_agent_login_type()
|| !ast_channel_agent_logoff_type()) {
unload_module();
return AST_MODULE_LOAD_DECLINE;
}
agent_router = stasis_message_router_create(ast_channel_topic_all()); agent_router = stasis_message_router_create(ast_channel_topic_all());
if (!agent_router) { if (!agent_router) {
unload_module();
return AST_MODULE_LOAD_DECLINE; return AST_MODULE_LOAD_DECLINE;
} }
err |= stasis_message_router_add(agent_router,
STASIS_MESSAGE_TYPE_INIT(queue_caller_join_type);
STASIS_MESSAGE_TYPE_INIT(queue_caller_leave_type);
STASIS_MESSAGE_TYPE_INIT(queue_caller_abandon_type);
STASIS_MESSAGE_TYPE_INIT(queue_member_status_type);
STASIS_MESSAGE_TYPE_INIT(queue_member_added_type);
STASIS_MESSAGE_TYPE_INIT(queue_member_removed_type);
STASIS_MESSAGE_TYPE_INIT(queue_member_pause_type);
STASIS_MESSAGE_TYPE_INIT(queue_member_penalty_type);
STASIS_MESSAGE_TYPE_INIT(queue_member_ringinuse_type);
STASIS_MESSAGE_TYPE_INIT(queue_agent_called_type);
STASIS_MESSAGE_TYPE_INIT(queue_agent_connect_type);
STASIS_MESSAGE_TYPE_INIT(queue_agent_complete_type);
STASIS_MESSAGE_TYPE_INIT(queue_agent_dump_type);
STASIS_MESSAGE_TYPE_INIT(queue_agent_ringnoanswer_type);
stasis_message_router_add(manager_router,
queue_caller_join_type(),
queue_channel_manager_event,
"QueueCallerJoin");
stasis_message_router_add(manager_router,
queue_caller_leave_type(),
queue_channel_manager_event,
"QueueCallerLeave");
stasis_message_router_add(manager_router,
queue_caller_abandon_type(),
queue_channel_manager_event,
"QueueCallerAbandon");
stasis_message_router_add(manager_router,
queue_member_status_type(),
queue_member_manager_event,
"QueueMemberStatus");
stasis_message_router_add(manager_router,
queue_member_added_type(),
queue_member_manager_event,
"QueueMemberAdded");
stasis_message_router_add(manager_router,
queue_member_removed_type(),
queue_member_manager_event,
"QueueMemberRemoved");
stasis_message_router_add(manager_router,
queue_member_pause_type(),
queue_member_manager_event,
"QueueMemberPause");
stasis_message_router_add(manager_router,
queue_member_penalty_type(),
queue_member_manager_event,
"QueueMemberPenalty");
stasis_message_router_add(manager_router,
queue_member_ringinuse_type(),
queue_member_manager_event,
"QueueMemberRinginuse");
stasis_message_router_add(manager_router,
queue_agent_called_type(),
queue_multi_channel_manager_event,
"AgentCalled");
stasis_message_router_add(manager_router,
queue_agent_connect_type(),
queue_multi_channel_manager_event,
"AgentConnect");
stasis_message_router_add(manager_router,
queue_agent_complete_type(),
queue_multi_channel_manager_event,
"AgentComplete");
stasis_message_router_add(manager_router,
queue_agent_dump_type(),
queue_multi_channel_manager_event,
"AgentDump");
stasis_message_router_add(manager_router,
queue_agent_ringnoanswer_type(),
queue_multi_channel_manager_event,
"AgentRingNoAnswer");
stasis_message_router_add(agent_router,
ast_channel_agent_login_type(), ast_channel_agent_login_type(),
queue_agent_cb, queue_agent_cb,
NULL); NULL);
err |= stasis_message_router_add(agent_router,
stasis_message_router_add(agent_router,
ast_channel_agent_logoff_type(), ast_channel_agent_logoff_type(),
queue_agent_cb, queue_agent_cb,
NULL); NULL);
err |= STASIS_MESSAGE_TYPE_INIT(queue_caller_join_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_caller_leave_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_caller_abandon_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_member_status_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_member_added_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_member_removed_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_member_pause_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_member_penalty_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_member_ringinuse_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_agent_called_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_agent_connect_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_agent_complete_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_agent_dump_type);
err |= STASIS_MESSAGE_TYPE_INIT(queue_agent_ringnoanswer_type);
ast_extension_state_add(NULL, NULL, extension_state_cb, NULL); ast_extension_state_add(NULL, NULL, extension_state_cb, NULL);
return res ? AST_MODULE_LOAD_DECLINE : 0; if (err) {
unload_module();
return AST_MODULE_LOAD_DECLINE;
}
return AST_MODULE_LOAD_SUCCESS;
} }
static int reload(void) static int reload(void)

156
res/res_agi.c
Unescape View File

@ -1033,38 +1033,66 @@ enum agi_result {
AGI_RESULT_HANGUP, AGI_RESULT_HANGUP,
}; };
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_exec_start_type); static struct ast_manager_event_blob *agi_channel_to_ami(const char *type, struct stasis_message *message)
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_exec_end_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_async_start_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_async_exec_type);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_async_end_type);
static void agi_channel_manager_event(void *data,
struct stasis_subscription *sub,
struct stasis_message *message)
{ {
const char *type = data;
struct ast_channel_blob *obj = stasis_message_data(message); struct ast_channel_blob *obj = stasis_message_data(message);
RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, channel_string, NULL, ast_free);
RAII_VAR(struct ast_str *, event_string, NULL, ast_free); RAII_VAR(struct ast_str *, event_string, NULL, ast_free);
channel_event_string = ast_manager_build_channel_state_string(obj->snapshot); channel_string = ast_manager_build_channel_state_string(obj->snapshot);
if (!channel_event_string) {
return;
}
event_string = ast_manager_str_from_json_object(obj->blob, NULL); event_string = ast_manager_str_from_json_object(obj->blob, NULL);
if (!event_string) { if (!channel_string || !event_string) {
return; return NULL;
} }
manager_event(EVENT_FLAG_AGI, type, return ast_manager_event_blob_create(EVENT_FLAG_AGI, type,
"%s" "%s"
"%s", "%s",
ast_str_buffer(channel_event_string), ast_str_buffer(channel_string),
ast_str_buffer(event_string)); ast_str_buffer(event_string));
} }
static struct ast_manager_event_blob *agi_exec_start_to_ami(struct stasis_message *message)
{
return agi_channel_to_ami("AGIExecStart", message);
}
static struct ast_manager_event_blob *agi_exec_end_to_ami(struct stasis_message *message)
{
return agi_channel_to_ami("AGIExecEnd", message);
}
static struct ast_manager_event_blob *agi_async_start_to_ami(struct stasis_message *message)
{
return agi_channel_to_ami("AsyncAGIStart", message);
}
static struct ast_manager_event_blob *agi_async_exec_to_ami(struct stasis_message *message)
{
return agi_channel_to_ami("AsyncAGIExec", message);
}
static struct ast_manager_event_blob *agi_async_end_to_ami(struct stasis_message *message)
{
return agi_channel_to_ami("AsyncAGIEnd", message);
}
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_exec_start_type,
.to_ami = agi_exec_start_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_exec_end_type,
.to_ami = agi_exec_end_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_async_start_type,
.to_ami = agi_async_start_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_async_exec_type,
.to_ami = agi_async_exec_to_ami,
);
STASIS_MESSAGE_TYPE_DEFN_LOCAL(agi_async_end_type,
.to_ami = agi_async_end_to_ami,
);
static agi_command *find_command(const char * const cmds[], int exact); static agi_command *find_command(const char * const cmds[], int exact);
AST_THREADSTORAGE(agi_buf); AST_THREADSTORAGE(agi_buf);
@ -3468,10 +3496,9 @@ int AST_OPTIONAL_API_NAME(ast_agi_unregister)(struct ast_module *mod, agi_comman
} }
AST_RWLIST_TRAVERSE_SAFE_END; AST_RWLIST_TRAVERSE_SAFE_END;
AST_RWLIST_UNLOCK(&agi_commands); AST_RWLIST_UNLOCK(&agi_commands);
if (unregistered) if (unregistered) {
ast_verb(2, "AGI Command '%s' unregistered\n",fullcmd); ast_verb(2, "AGI Command '%s' unregistered\n",fullcmd);
else }
ast_log(LOG_WARNING, "Unable to unregister command: '%s'!\n",fullcmd);
return unregistered; return unregistered;
} }
@ -4256,17 +4283,6 @@ AST_TEST_DEFINE(test_agi_null_docs)
static int unload_module(void) static int unload_module(void)
{ {
struct stasis_message_router *message_router;
message_router = ast_manager_get_message_router();
if (message_router) {
stasis_message_router_remove(message_router, agi_exec_start_type());
stasis_message_router_remove(message_router, agi_exec_end_type());
stasis_message_router_remove(message_router, agi_async_start_type());
stasis_message_router_remove(message_router, agi_async_exec_type());
stasis_message_router_remove(message_router, agi_async_end_type());
}
STASIS_MESSAGE_TYPE_CLEANUP(agi_exec_start_type); STASIS_MESSAGE_TYPE_CLEANUP(agi_exec_start_type);
STASIS_MESSAGE_TYPE_CLEANUP(agi_exec_end_type); STASIS_MESSAGE_TYPE_CLEANUP(agi_exec_end_type);
STASIS_MESSAGE_TYPE_CLEANUP(agi_async_start_type); STASIS_MESSAGE_TYPE_CLEANUP(agi_async_start_type);
@ -4274,67 +4290,39 @@ static int unload_module(void)
STASIS_MESSAGE_TYPE_CLEANUP(agi_async_end_type); STASIS_MESSAGE_TYPE_CLEANUP(agi_async_end_type);
ast_cli_unregister_multiple(cli_agi, ARRAY_LEN(cli_agi)); ast_cli_unregister_multiple(cli_agi, ARRAY_LEN(cli_agi));
/* we can safely ignore the result of ast_agi_unregister_multiple() here, since it cannot fail, as ast_agi_unregister_multiple(ast_module_info->self, commands, ARRAY_LEN(commands));
we know that these commands were registered by this module and are still registered
*/
(void) ast_agi_unregister_multiple(ast_module_info->self, commands, ARRAY_LEN(commands));
ast_unregister_application(eapp); ast_unregister_application(eapp);
ast_unregister_application(deadapp); ast_unregister_application(deadapp);
ast_manager_unregister("AGI"); ast_manager_unregister("AGI");
ast_unregister_application(app);
AST_TEST_UNREGISTER(test_agi_null_docs); AST_TEST_UNREGISTER(test_agi_null_docs);
return ast_unregister_application(app); return 0;
} }
static int load_module(void) static int load_module(void)
{ {
struct stasis_message_router *message_router; int err = 0;
message_router = ast_manager_get_message_router(); err |= STASIS_MESSAGE_TYPE_INIT(agi_exec_start_type);
if (!message_router) { err |= STASIS_MESSAGE_TYPE_INIT(agi_exec_end_type);
return AST_MODULE_LOAD_DECLINE; err |= STASIS_MESSAGE_TYPE_INIT(agi_async_start_type);
} err |= STASIS_MESSAGE_TYPE_INIT(agi_async_exec_type);
err |= STASIS_MESSAGE_TYPE_INIT(agi_async_end_type);
err |= ast_cli_register_multiple(cli_agi, ARRAY_LEN(cli_agi));
err |= ast_agi_register_multiple(ast_module_info->self, commands, ARRAY_LEN(commands));
err |= ast_register_application_xml(deadapp, deadagi_exec);
err |= ast_register_application_xml(eapp, eagi_exec);
err |= ast_manager_register_xml("AGI", EVENT_FLAG_AGI, action_add_agi_cmd);
err |= ast_register_application_xml(app, agi_exec);
STASIS_MESSAGE_TYPE_INIT(agi_exec_start_type);
STASIS_MESSAGE_TYPE_INIT(agi_exec_end_type);
STASIS_MESSAGE_TYPE_INIT(agi_async_start_type);
STASIS_MESSAGE_TYPE_INIT(agi_async_exec_type);
STASIS_MESSAGE_TYPE_INIT(agi_async_end_type);
stasis_message_router_add(message_router,
agi_exec_start_type(),
agi_channel_manager_event,
"AGIExecStart");
stasis_message_router_add(message_router,
agi_exec_end_type(),
agi_channel_manager_event,
"AGIExecEnd");
stasis_message_router_add(message_router,
agi_async_start_type(),
agi_channel_manager_event,
"AsyncAGIStart");
stasis_message_router_add(message_router,
agi_async_exec_type(),
agi_channel_manager_event,
"AsyncAGIExec");
stasis_message_router_add(message_router,
agi_async_end_type(),
agi_channel_manager_event,
"AsyncAGIEnd");
ast_cli_register_multiple(cli_agi, ARRAY_LEN(cli_agi));
/* we can safely ignore the result of ast_agi_register_multiple() here, since it cannot fail, as
no other commands have been registered yet
*/
(void) ast_agi_register_multiple(ast_module_info->self, commands, ARRAY_LEN(commands));
ast_register_application_xml(deadapp, deadagi_exec);
ast_register_application_xml(eapp, eagi_exec);
ast_manager_register_xml("AGI", EVENT_FLAG_AGI, action_add_agi_cmd);
AST_TEST_REGISTER(test_agi_null_docs); AST_TEST_REGISTER(test_agi_null_docs);
return ast_register_application_xml(app, agi_exec);
if (err) {
unload_module();
return AST_MODULE_LOAD_DECLINE;
}
return AST_MODULE_LOAD_SUCCESS;
} }
AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS | AST_MODFLAG_LOAD_ORDER, "Asterisk Gateway Interface (AGI)", AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS | AST_MODFLAG_LOAD_ORDER, "Asterisk Gateway Interface (AGI)",

Write Preview
Loading…
Cancel
Save