the message will automatically be associated with the configured endpoint on the
outbound registration.
+
+Core
+------------------
+ * The core of Asterisk uses a message bus called "Stasis" to distribute
+ information to internal components. For performance reasons, the message
+ distribution was modified to make use of a thread pool instead of a
+ dedicated thread per consumer in certain cases. The initial settings for
+ the thread pool can now be configured in 'stasis.conf'.
+
+
Functions
------------------
* AMI action PJSIPNotify may now send to a URI instead of only to a PJSIP
endpoint as long as a default outbound endpoint is set. This also applies
to the equivalent CLI command (pjsip send notify)
-
+
* The AMI action PJSIPShowEndpoint now includes ContactStatusDetail sections
that give information on Asterisk's attempts to qualify the endpoint.
return -1;
}
- queue_data->bridge_router = stasis_message_router_create(ast_bridge_topic_all());
+ queue_data->bridge_router = stasis_message_router_create_pool(ast_bridge_topic_all());
if (!queue_data->bridge_router) {
ao2_ref(queue_data, -1);
return -1;
stasis_message_router_set_default(queue_data->bridge_router,
queue_bridge_cb, queue_data);
- queue_data->channel_router = stasis_message_router_create(ast_channel_topic_all());
+ queue_data->channel_router = stasis_message_router_create_pool(ast_channel_topic_all());
if (!queue_data->channel_router) {
/* Unsubscribing from the bridge router will remove the only ref of queue_data,
* thus beginning the destruction process
mailbox_specific_topic = ast_mwi_topic(tmp->mailbox);
if (mailbox_specific_topic) {
- tmp->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, NULL);
+ tmp->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, NULL);
}
}
#ifdef HAVE_DAHDI_LINEREVERSE_VMWI
mailbox_specific_topic = ast_mwi_topic(peer->mailbox);
if (mailbox_specific_topic) {
- peer->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, NULL);
+ peer->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, NULL);
}
}
mailbox_specific_topic = ast_mwi_topic(e->mailbox);
if (mailbox_specific_topic) {
- e->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, NULL);
+ e->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, NULL);
}
}
snprintf(e->rqnt_ident, sizeof(e->rqnt_ident), "%08lx", (unsigned long)ast_random());
if (!peer_name) {
return;
}
- mailbox->event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, peer_name);
+ mailbox->event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, peer_name);
}
}
}
mailbox_specific_topic = ast_mwi_topic(l->mailbox);
if (mailbox_specific_topic) {
- l->mwi_event_sub = stasis_subscribe(mailbox_specific_topic, mwi_event_cb, l);
+ l->mwi_event_sub = stasis_subscribe_pool(mailbox_specific_topic, mwi_event_cb, l);
}
}
mailbox_specific_topic = ast_mwi_topic(mbox_id);
if (mailbox_specific_topic) {
- pri->mbox[i].sub = stasis_subscribe(mailbox_specific_topic, sig_pri_mwi_event_cb, pri);
+ pri->mbox[i].sub = stasis_subscribe_pool(mailbox_specific_topic, sig_pri_mwi_event_cb, pri);
}
if (!pri->mbox[i].sub) {
ast_log(LOG_ERROR, "%s span %d could not subscribe to MWI events for %s(%s).\n",
+[threadpool]
+;initial_size = 5 ; Initial size of the threadpool.
+; ; 0 means the threadpool has no threads initially
+; ; until a task needs a thread.
+;idle_timeout_sec = 20 ; Number of seconds a thread should be idle before
+; ; dying. 0 means threads never time out.
+;max_size = 50 ; Maximum number of threads in the Stasis threadpool.
+; ; 0 means no limit to the number of threads in the
+; ; threadpool.
+
[declined_message_types]
; This config section contains the names of message types that should be prevented
; from being created. By default, all message types are allowed to be created.
stasis_subscription_cb callback, void *data);
/*!
+ * \brief Create a subscription whose callbacks occur on a thread pool
+ *
+ * In addition to being AO2 managed memory (requiring an ao2_cleanup() to free
+ * up this reference), the subscription must be explicitly unsubscribed from its
+ * topic using stasis_unsubscribe().
+ *
+ * The invocations of the callback are serialized, but will almost certainly not
+ * always happen on the same thread. The invocation order of different subscriptions
+ * is unspecified.
+ *
+ * Unlike \ref stasis_subscribe, this function will explicitly use a threadpool to
+ * dispatch items to its \c callback. This form of subscription should be used
+ * when many subscriptions may be made to the specified \c topic.
+ *
+ * \param topic Topic to subscribe to.
+ * \param callback Callback function for subscription messages.
+ * \param data Data to be passed to the callback, in addition to the message.
+ * \return New \ref stasis_subscription object.
+ * \return \c NULL on error.
+ * \since 12.8.0
+ */
+struct stasis_subscription *stasis_subscribe_pool(struct stasis_topic *topic,
+ stasis_subscription_cb callback, void *data);
+
+/*!
* \brief Cancel a subscription.
*
* Note that in an asynchronous system, there may still be messages queued or
* \param callback Callback function for subscription messages.
* \param data Data to be passed to the callback, in addition to the message.
* \param needs_mailbox Determines whether or not the subscription requires a mailbox.
- * Subscriptions with mailboxes will be delivered on a thread in the Stasis threadpool;
+ * Subscriptions with mailboxes will be delivered on some non-publisher thread;
* subscriptions without mailboxes will be delivered on the publisher thread.
+ * \param use_thread_pool Use the thread pool for the subscription. This is only
+ * relevant if \c needs_mailbox is non-zero.
* \return New \ref stasis_subscription object.
* \return \c NULL on error.
* \since 12
struct stasis_topic *topic,
stasis_subscription_cb callback,
void *data,
- int needs_mailbox);
+ int needs_mailbox,
+ int use_thread_pool);
#endif /* STASIS_INTERNAL_H_ */
struct stasis_topic *topic);
/*!
+ * \brief Create a new message router object.
+ *
+ * The subscription created for this message router will dispatch
+ * callbacks on a thread pool.
+ *
+ * \param topic Topic to subscribe route to.
+ *
+ * \return New \ref stasis_message_router.
+ * \return \c NULL on error.
+ *
+ * \since 12.8.0
+ */
+struct stasis_message_router *stasis_message_router_create_pool(
+ struct stasis_topic *topic);
+
+/*!
* \brief Unsubscribe the router from the upstream topic.
*
* \param router Router to unsubscribe.
}
if (!ast_strlen_zero(resource)) {
- endpoint->router = stasis_message_router_create(ast_endpoint_topic(endpoint));
+ endpoint->router = stasis_message_router_create_pool(ast_endpoint_topic(endpoint));
if (!endpoint->router) {
return NULL;
}
#include "asterisk/stasis_internal.h"
#include "asterisk/stasis.h"
#include "asterisk/taskprocessor.h"
+#include "asterisk/threadpool.h"
#include "asterisk/utils.h"
#include "asterisk/uuid.h"
#include "asterisk/vector.h"
</managerEvent>
<configInfo name="stasis" language="en_US">
<configFile name="stasis.conf">
+ <configObject name="threadpool">
+ <synopsis>Settings that configure the threadpool Stasis uses to deliver some messages.</synopsis>
+ <configOption name="initial_size" default="5">
+ <synopsis>Initial number of threads in the message bus threadpool.</synopsis>
+ </configOption>
+ <configOption name="idle_timeout_sec" default="20">
+ <synopsis>Number of seconds before an idle thread is disposed of.</synopsis>
+ </configOption>
+ <configOption name="max_size" default="50">
+ <synopsis>Maximum number of threads in the threadpool.</synopsis>
+ </configOption>
+ </configObject>
<configObject name="declined_message_types">
<synopsis>Stasis message types for which to decline creation.</synopsis>
<configOption name="decline">
/*! The number of buckets to use for topic pools */
#define TOPIC_POOL_BUCKETS 57
+/*! Thread pool for topics that don't want a dedicated taskprocessor */
+static struct ast_threadpool *pool;
+
STASIS_MESSAGE_TYPE_DEFN(stasis_subscription_change_type);
/*! \internal */
struct stasis_topic *topic,
stasis_subscription_cb callback,
void *data,
- int needs_mailbox)
+ int needs_mailbox,
+ int use_thread_pool)
{
RAII_VAR(struct stasis_subscription *, sub, NULL, ao2_cleanup);
if (!sub) {
return NULL;
}
-
ast_uuid_generate_str(sub->uniqueid, sizeof(sub->uniqueid));
if (needs_mailbox) {
/* With a small number of subscribers, a thread-per-sub is
- * acceptable. If our usage changes so that we have larger
- * numbers of subscribers, we'll probably want to consider
- * a threadpool. We had that originally, but with so few
- * subscribers it was actually a performance loss instead of
- * a gain.
+ * acceptable. For larger number of subscribers, a thread
+ * pool should be used.
*/
- sub->mailbox = ast_taskprocessor_get(sub->uniqueid,
- TPS_REF_DEFAULT);
+ if (use_thread_pool) {
+ sub->mailbox = ast_threadpool_serializer(sub->uniqueid, pool);
+ } else {
+ sub->mailbox = ast_taskprocessor_get(sub->uniqueid,
+ TPS_REF_DEFAULT);
+ }
if (!sub->mailbox) {
return NULL;
}
stasis_subscription_cb callback,
void *data)
{
- return internal_stasis_subscribe(topic, callback, data, 1);
+ return internal_stasis_subscribe(topic, callback, data, 1, 0);
+}
+
+struct stasis_subscription *stasis_subscribe_pool(
+ struct stasis_topic *topic,
+ stasis_subscription_cb callback,
+ void *data)
+{
+ return internal_stasis_subscribe(topic, callback, data, 1, 1);
}
static int sub_cleanup(void *data)
struct ao2_container *declined;
};
+/*! \brief Threadpool configuration options */
+struct stasis_threadpool_conf {
+ /*! Initial size of the thread pool */
+ int initial_size;
+ /*! Time, in seconds, before we expire a thread */
+ int idle_timeout_sec;
+ /*! Maximum number of thread to allow */
+ int max_size;
+};
struct stasis_config {
+ /*! Thread pool configuration options */
+ struct stasis_threadpool_conf *threadpool_options;
+ /*! Declined message types */
struct stasis_declined_config *declined_message_types;
};
+static struct aco_type threadpool_option = {
+ .type = ACO_GLOBAL,
+ .name = "threadpool",
+ .item_offset = offsetof(struct stasis_config, threadpool_options),
+ .category = "^threadpool$",
+ .category_match = ACO_WHITELIST,
+};
+
+static struct aco_type *threadpool_options[] = ACO_TYPES(&threadpool_option);
+
/*! \brief An aco_type structure to link the "declined_message_types" category to the stasis_declined_config type */
static struct aco_type declined_option = {
.type = ACO_GLOBAL,
struct aco_file stasis_conf = {
.filename = "stasis.conf",
- .types = ACO_TYPES(&declined_option),
+ .types = ACO_TYPES(&declined_option, &threadpool_option),
};
/*! \brief A global object container that will contain the stasis_config that gets swapped out on reloads */
static void stasis_declined_config_destructor(void *obj)
{
struct stasis_declined_config *declined = obj;
+
ao2_cleanup(declined->declined);
}
static void stasis_config_destructor(void *obj)
{
struct stasis_config *cfg = obj;
+
ao2_cleanup(cfg->declined_message_types);
+ ast_free(cfg->threadpool_options);
}
static void *stasis_config_alloc(void)
return NULL;
}
- /* Allocate/initialize memory */
- cfg->declined_message_types = ao2_alloc(sizeof(*cfg->declined_message_types), stasis_declined_config_destructor);
+ cfg->threadpool_options = ast_calloc(1, sizeof(*cfg->threadpool_options));
+ if (!cfg->threadpool_options) {
+ ao2_ref(cfg, -1);
+ return NULL;
+ }
+
+ cfg->declined_message_types = ao2_alloc(sizeof(*cfg->declined_message_types),
+ stasis_declined_config_destructor);
if (!cfg->declined_message_types) {
- goto error;
+ ao2_ref(cfg, -1);
+ return NULL;
}
cfg->declined_message_types->declined = ast_str_container_alloc(13);
if (!cfg->declined_message_types->declined) {
- goto error;
+ ao2_ref(cfg, -1);
+ return NULL;
}
return cfg;
-error:
- ao2_ref(cfg, -1);
- return NULL;
}
int stasis_message_type_declined(const char *name)
/*! @} */
+/*! \brief Shutdown function */
+static void stasis_exit(void)
+{
+ ast_threadpool_shutdown(pool);
+ pool = NULL;
+}
+
/*! \brief Cleanup function for graceful shutdowns */
static void stasis_cleanup(void)
{
int stasis_init(void)
{
+ RAII_VAR(struct stasis_config *, cfg, NULL, ao2_cleanup);
int cache_init;
+ struct ast_threadpool_options threadpool_opts = { 0, };
/* Be sure the types are cleaned up after the message bus */
ast_register_cleanup(stasis_cleanup);
+ ast_register_atexit(stasis_exit);
if (aco_info_init(&cfg_info)) {
return -1;
}
- aco_option_register_custom(&cfg_info, "decline", ACO_EXACT, declined_options, "", declined_handler, 0);
+ aco_option_register_custom(&cfg_info, "decline", ACO_EXACT,
+ declined_options, "", declined_handler, 0);
+ aco_option_register(&cfg_info, "initial_size", ACO_EXACT,
+ threadpool_options, "5", OPT_INT_T, PARSE_IN_RANGE,
+ FLDSET(struct stasis_threadpool_conf, initial_size), 0,
+ INT_MAX);
+ aco_option_register(&cfg_info, "idle_timeout_sec", ACO_EXACT,
+ threadpool_options, "20", OPT_INT_T, PARSE_IN_RANGE,
+ FLDSET(struct stasis_threadpool_conf, idle_timeout_sec), 0,
+ INT_MAX);
+ aco_option_register(&cfg_info, "max_size", ACO_EXACT,
+ threadpool_options, "50", OPT_INT_T, PARSE_IN_RANGE,
+ FLDSET(struct stasis_threadpool_conf, max_size), 0,
+ INT_MAX);
if (aco_process_config(&cfg_info, 0) == ACO_PROCESS_ERROR) {
- RAII_VAR(struct stasis_config *, stasis_cfg, stasis_config_alloc(), ao2_cleanup);
+ struct stasis_config *default_cfg = stasis_config_alloc();
+
+ if (!default_cfg) {
+ return -1;
+ }
- if (aco_set_defaults(&declined_option, "declined_message_types", stasis_cfg->declined_message_types)) {
+ if (aco_set_defaults(&threadpool_option, "threadpool", default_cfg->threadpool_options)) {
+ ast_log(LOG_ERROR, "Failed to initialize defaults on Stasis configuration object\n");
+ ao2_ref(default_cfg, -1);
+ return -1;
+ }
+
+ if (aco_set_defaults(&declined_option, "declined_message_types", default_cfg->declined_message_types)) {
ast_log(LOG_ERROR, "Failed to load stasis.conf and failed to initialize defaults.\n");
return -1;
}
- ast_log(LOG_NOTICE, "Could not load stasis config; using defaults\n");
- ao2_global_obj_replace_unref(globals, stasis_cfg);
+ ast_log(LOG_NOTICE, "Could not load Stasis configuration; using defaults\n");
+ ao2_global_obj_replace_unref(globals, default_cfg);
+ cfg = default_cfg;
+ } else {
+ cfg = ao2_global_obj_ref(globals);
+ if (!cfg) {
+ ast_log(LOG_ERROR, "Failed to obtain Stasis configuration object\n");
+ return -1;
+ }
+ }
+
+ threadpool_opts.version = AST_THREADPOOL_OPTIONS_VERSION;
+ threadpool_opts.initial_size = cfg->threadpool_options->initial_size;
+ threadpool_opts.auto_increment = 1;
+ threadpool_opts.max_size = cfg->threadpool_options->max_size;
+ threadpool_opts.idle_timeout = cfg->threadpool_options->idle_timeout_sec;
+ pool = ast_threadpool_create("stasis-core", NULL, &threadpool_opts);
+ if (!pool) {
+ ast_log(LOG_ERROR, "Failed to create 'stasis-core' threadpool\n");
+ return -1;
}
cache_init = stasis_cache_init();
ao2_ref(cache, +1);
caching_topic->cache = cache;
- sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0);
+ sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0, 0);
if (sub == NULL) {
return NULL;
}
}
}
-struct stasis_message_router *stasis_message_router_create(
- struct stasis_topic *topic)
+static struct stasis_message_router *stasis_message_router_create_internal(
+ struct stasis_topic *topic, int use_thread_pool)
{
int res;
RAII_VAR(struct stasis_message_router *, router, NULL, ao2_cleanup);
return NULL;
}
- router->subscription = stasis_subscribe(topic, router_dispatch, router);
+ if (use_thread_pool) {
+ router->subscription = stasis_subscribe_pool(topic, router_dispatch, router);
+ } else {
+ router->subscription = stasis_subscribe(topic, router_dispatch, router);
+ }
if (!router->subscription) {
return NULL;
}
return router;
}
+struct stasis_message_router *stasis_message_router_create(
+ struct stasis_topic *topic)
+{
+ return stasis_message_router_create_internal(topic, 0);
+}
+
+struct stasis_message_router *stasis_message_router_create_pool(
+ struct stasis_topic *topic)
+{
+ return stasis_message_router_create_internal(topic, 1);
+}
+
void stasis_message_router_unsubscribe(struct stasis_message_router *router)
{
if (!router) {
return -1;
}
- if (!(parking_subscription = stasis_subscribe(ast_parking_topic(), park_announce_update_cb, pa_data))) {
+ if (!(parking_subscription = stasis_subscribe_pool(ast_parking_topic(), park_announce_update_cb, pa_data))) {
/* Failed to create subscription */
park_announce_subscription_data_destroy(pa_data);
return -1;
strcpy(subscription_data->parkee_uuid, parkee_uuid);
strcpy(subscription_data->parker_uuid, parker_uuid);
- if (!(parked_datastore->parked_subscription = stasis_subscribe(ast_parking_topic(), parker_update_cb, subscription_data))) {
+ if (!(parked_datastore->parked_subscription = stasis_subscribe_pool(ast_parking_topic(), parker_update_cb, subscription_data))) {
return -1;
}
strcpy(mwi_stasis_sub->mailbox, mailbox);
ao2_ref(mwi_sub, +1);
ast_debug(3, "Creating stasis MWI subscription to mailbox %s for endpoint %s\n", mailbox, mwi_sub->id);
- mwi_stasis_sub->stasis_sub = stasis_subscribe(topic, mwi_stasis_cb, mwi_sub);
+ mwi_stasis_sub->stasis_sub = stasis_subscribe_pool(topic, mwi_stasis_cb, mwi_sub);
return mwi_stasis_sub;
}
if (ast_test_flag(&ast_options, AST_OPT_FLAG_FULLY_BOOTED)) {
ast_sip_push_task(NULL, subscription_persistence_load, NULL);
} else {
- stasis_subscribe(ast_manager_get_topic(), subscription_persistence_event_cb, NULL);
+ stasis_subscribe_pool(ast_manager_get_topic(), subscription_persistence_event_cb, NULL);
}
ast_manager_register_xml(AMI_SHOW_SUBSCRIPTIONS_INBOUND, EVENT_FLAG_SYSTEM,
/* We also will need to detect if the transferee enters a bridge. This is currently the only reliable way to
* detect if the transfer target has answered the call
*/
- refer->progress->bridge_sub = stasis_subscribe(ast_bridge_topic_all(), refer_progress_bridge, refer->progress);
+ refer->progress->bridge_sub = stasis_subscribe_pool(ast_bridge_topic_all(), refer_progress_bridge, refer->progress);
if (!refer->progress->bridge_sub) {
struct refer_progress_notification *notification = refer_progress_notification_alloc(refer->progress, 200,
PJSIP_EVSUB_STATE_TERMINATED);
return 0;
}
- if (!(sub->sub = stasis_subscribe(
+ if (!(sub->sub = stasis_subscribe_pool(
ast_device_state_topic(sub->device_name),
device_state_cb, sub))) {
ast_log(LOG_ERROR, "Unable to subscribe to device %s\n",
xmpp_pubsub_unsubscribe(client, "device_state");
xmpp_pubsub_unsubscribe(client, "message_waiting");
- if (!(client->mwi_sub = stasis_subscribe(ast_mwi_topic_all(), xmpp_pubsub_mwi_cb, client))) {
+ if (!(client->mwi_sub = stasis_subscribe_pool(ast_mwi_topic_all(), xmpp_pubsub_mwi_cb, client))) {
return;
}
return AST_TEST_PASS;
}
+AST_TEST_DEFINE(subscription_pool_messages)
+{
+ RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
+ RAII_VAR(char *, test_data, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
+ RAII_VAR(char *, expected_uniqueid, NULL, ast_free);
+ int complete;
+ struct stasis_subscription_change *change;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = __func__;
+ info->category = test_category;
+ info->summary = "Test subscribe/unsubscribe messages using a threadpool subscription";
+ info->description = "Test subscribe/unsubscribe messages using a threadpool subscription";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ topic = stasis_topic_create("TestTopic");
+ ast_test_validate(test, NULL != topic);
+
+ consumer = consumer_create(0);
+ ast_test_validate(test, NULL != consumer);
+
+ uut = stasis_subscribe_pool(topic, consumer_exec, consumer);
+ ast_test_validate(test, NULL != uut);
+ ao2_ref(consumer, +1);
+ expected_uniqueid = ast_strdup(stasis_subscription_uniqueid(uut));
+
+ uut = stasis_unsubscribe(uut);
+ complete = consumer_wait_for_completion(consumer);
+ ast_test_validate(test, 1 == complete);
+
+ ast_test_validate(test, 2 == consumer->messages_rxed_len);
+ ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[0]));
+ ast_test_validate(test, stasis_subscription_change_type() == stasis_message_type(consumer->messages_rxed[1]));
+
+ change = stasis_message_data(consumer->messages_rxed[0]);
+ ast_test_validate(test, topic == change->topic);
+ ast_test_validate(test, 0 == strcmp("Subscribe", change->description));
+ ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
+
+ change = stasis_message_data(consumer->messages_rxed[1]);
+ ast_test_validate(test, topic == change->topic);
+ ast_test_validate(test, 0 == strcmp("Unsubscribe", change->description));
+ ast_test_validate(test, 0 == strcmp(expected_uniqueid, change->uniqueid));
+
+ return AST_TEST_PASS;
+}
+
AST_TEST_DEFINE(publish)
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
return AST_TEST_PASS;
}
+AST_TEST_DEFINE(publish_pool)
+{
+ RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
+ RAII_VAR(char *, test_data, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
+ int actual_len;
+ const char *actual;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = __func__;
+ info->category = test_category;
+ info->summary = "Test publishing with a threadpool";
+ info->description = "Test publishing to a subscriber whose\n"
+ "subscription dictates messages are received through a\n"
+ "threadpool.";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ topic = stasis_topic_create("TestTopic");
+ ast_test_validate(test, NULL != topic);
+
+ consumer = consumer_create(1);
+ ast_test_validate(test, NULL != consumer);
+
+ uut = stasis_subscribe_pool(topic, consumer_exec, consumer);
+ ast_test_validate(test, NULL != uut);
+ ao2_ref(consumer, +1);
+
+ test_data = ao2_alloc(1, NULL);
+ ast_test_validate(test, NULL != test_data);
+ test_message_type = stasis_message_type_create("TestMessage", NULL);
+ test_message = stasis_message_create(test_message_type, test_data);
+
+ stasis_publish(topic, test_message);
+
+ actual_len = consumer_wait_for(consumer, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual = stasis_message_data(consumer->messages_rxed[0]);
+ ast_test_validate(test, test_data == actual);
+
+ return AST_TEST_PASS;
+}
+
AST_TEST_DEFINE(unsubscribe_stops_messages)
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
return AST_TEST_PASS;
}
+AST_TEST_DEFINE(subscription_interleaving)
+{
+ RAII_VAR(struct stasis_topic *, parent_topic, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_topic *, topic1, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_topic *, topic2, NULL, ao2_cleanup);
+
+ RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
+
+ RAII_VAR(char *, test_data, NULL, ao2_cleanup);
+
+ RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
+
+ RAII_VAR(struct stasis_forward *, forward_sub1, NULL, stasis_forward_cancel);
+ RAII_VAR(struct stasis_forward *, forward_sub2, NULL, stasis_forward_cancel);
+ RAII_VAR(struct stasis_subscription *, sub1, NULL, stasis_unsubscribe);
+ RAII_VAR(struct stasis_subscription *, sub2, NULL, stasis_unsubscribe);
+
+ RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
+
+ int actual_len;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = __func__;
+ info->category = test_category;
+ info->summary = "Test sending interleaved events to a parent topic with different subscribers";
+ info->description = "Test sending events to a parent topic.\n"
+ "This test creates three topics (one parent, two children)\n"
+ "and publishes messages alternately between the children.\n"
+ "It verifies that the messages are received in the expected\n"
+ "order, for different subscription types: one with a dedicated\n"
+ "thread, the other on the Stasis threadpool.\n";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ test_message_type = stasis_message_type_create("test", NULL);
+ ast_test_validate(test, NULL != test_message_type);
+
+ test_data = ao2_alloc(1, NULL);
+ ast_test_validate(test, NULL != test_data);
+
+ test_message1 = stasis_message_create(test_message_type, test_data);
+ ast_test_validate(test, NULL != test_message1);
+ test_message2 = stasis_message_create(test_message_type, test_data);
+ ast_test_validate(test, NULL != test_message2);
+ test_message3 = stasis_message_create(test_message_type, test_data);
+ ast_test_validate(test, NULL != test_message3);
+
+ parent_topic = stasis_topic_create("ParentTestTopic");
+ ast_test_validate(test, NULL != parent_topic);
+ topic1 = stasis_topic_create("Topic1");
+ ast_test_validate(test, NULL != topic1);
+ topic2 = stasis_topic_create("Topic2");
+ ast_test_validate(test, NULL != topic2);
+
+ forward_sub1 = stasis_forward_all(topic1, parent_topic);
+ ast_test_validate(test, NULL != forward_sub1);
+ forward_sub2 = stasis_forward_all(topic2, parent_topic);
+ ast_test_validate(test, NULL != forward_sub2);
+
+ consumer1 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer1);
+
+ consumer2 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer2);
+
+ sub1 = stasis_subscribe(parent_topic, consumer_exec, consumer1);
+ ast_test_validate(test, NULL != sub1);
+ ao2_ref(consumer1, +1);
+
+ sub2 = stasis_subscribe_pool(parent_topic, consumer_exec, consumer2);
+ ast_test_validate(test, NULL != sub2);
+ ao2_ref(consumer2, +1);
+
+ stasis_publish(topic1, test_message1);
+ stasis_publish(topic2, test_message2);
+ stasis_publish(topic1, test_message3);
+
+ actual_len = consumer_wait_for(consumer1, 3);
+ ast_test_validate(test, 3 == actual_len);
+
+ actual_len = consumer_wait_for(consumer2, 3);
+ ast_test_validate(test, 3 == actual_len);
+
+ ast_test_validate(test, test_message1 == consumer1->messages_rxed[0]);
+ ast_test_validate(test, test_message2 == consumer1->messages_rxed[1]);
+ ast_test_validate(test, test_message3 == consumer1->messages_rxed[2]);
+
+ ast_test_validate(test, test_message1 == consumer2->messages_rxed[0]);
+ ast_test_validate(test, test_message2 == consumer2->messages_rxed[1]);
+ ast_test_validate(test, test_message3 == consumer2->messages_rxed[2]);
+
+ return AST_TEST_PASS;
+}
+
struct cache_test_data {
char *id;
char *value;
return AST_TEST_PASS;
}
+AST_TEST_DEFINE(router_pool)
+{
+ RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe_and_join);
+ RAII_VAR(char *, test_data, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_type *, test_message_type1, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_type *, test_message_type2, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_type *, test_message_type3, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer3, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message1, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message2, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, test_message3, NULL, ao2_cleanup);
+ int actual_len, ret;
+ struct stasis_message *actual;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = __func__;
+ info->category = test_category;
+ info->summary = "Test message routing via threadpool";
+ info->description = "Test simple message routing when\n"
+ "the subscriptions dictate usage of the Stasis\n"
+ "threadpool.\n";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ topic = stasis_topic_create("TestTopic");
+ ast_test_validate(test, NULL != topic);
+
+ consumer1 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer1);
+ consumer2 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer2);
+ consumer3 = consumer_create(1);
+ ast_test_validate(test, NULL != consumer3);
+
+ test_message_type1 = stasis_message_type_create("TestMessage1", NULL);
+ ast_test_validate(test, NULL != test_message_type1);
+ test_message_type2 = stasis_message_type_create("TestMessage2", NULL);
+ ast_test_validate(test, NULL != test_message_type2);
+ test_message_type3 = stasis_message_type_create("TestMessage3", NULL);
+ ast_test_validate(test, NULL != test_message_type3);
+
+ uut = stasis_message_router_create_pool(topic);
+ ast_test_validate(test, NULL != uut);
+
+ ret = stasis_message_router_add(
+ uut, test_message_type1, consumer_exec, consumer1);
+ ast_test_validate(test, 0 == ret);
+ ao2_ref(consumer1, +1);
+ ret = stasis_message_router_add(
+ uut, test_message_type2, consumer_exec, consumer2);
+ ast_test_validate(test, 0 == ret);
+ ao2_ref(consumer2, +1);
+ ret = stasis_message_router_set_default(uut, consumer_exec, consumer3);
+ ast_test_validate(test, 0 == ret);
+ ao2_ref(consumer3, +1);
+
+ test_data = ao2_alloc(1, NULL);
+ ast_test_validate(test, NULL != test_data);
+ test_message1 = stasis_message_create(test_message_type1, test_data);
+ ast_test_validate(test, NULL != test_message1);
+ test_message2 = stasis_message_create(test_message_type2, test_data);
+ ast_test_validate(test, NULL != test_message2);
+ test_message3 = stasis_message_create(test_message_type3, test_data);
+ ast_test_validate(test, NULL != test_message3);
+
+ stasis_publish(topic, test_message1);
+ stasis_publish(topic, test_message2);
+ stasis_publish(topic, test_message3);
+
+ actual_len = consumer_wait_for(consumer1, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual_len = consumer_wait_for(consumer2, 1);
+ ast_test_validate(test, 1 == actual_len);
+ actual_len = consumer_wait_for(consumer3, 1);
+ ast_test_validate(test, 1 == actual_len);
+
+ actual = consumer1->messages_rxed[0];
+ ast_test_validate(test, test_message1 == actual);
+
+ actual = consumer2->messages_rxed[0];
+ ast_test_validate(test, test_message2 == actual);
+
+ actual = consumer3->messages_rxed[0];
+ ast_test_validate(test, test_message3 == actual);
+
+ /* consumer1 and consumer2 do not get the final message. */
+ ao2_cleanup(consumer1);
+ ao2_cleanup(consumer2);
+
+ return AST_TEST_PASS;
+}
+
static const char *cache_simple(struct stasis_message *message)
{
const char *type_name =
AST_TEST_UNREGISTER(message_type);
AST_TEST_UNREGISTER(message);
AST_TEST_UNREGISTER(subscription_messages);
+ AST_TEST_UNREGISTER(subscription_pool_messages);
AST_TEST_UNREGISTER(publish);
AST_TEST_UNREGISTER(publish_sync);
+ AST_TEST_UNREGISTER(publish_pool);
AST_TEST_UNREGISTER(unsubscribe_stops_messages);
AST_TEST_UNREGISTER(forward);
AST_TEST_UNREGISTER(cache_filter);
AST_TEST_UNREGISTER(cache_dump);
AST_TEST_UNREGISTER(cache_eid_aggregate);
AST_TEST_UNREGISTER(router);
+ AST_TEST_UNREGISTER(router_pool);
AST_TEST_UNREGISTER(router_cache_updates);
AST_TEST_UNREGISTER(interleaving);
+ AST_TEST_UNREGISTER(subscription_interleaving);
AST_TEST_UNREGISTER(no_to_json);
AST_TEST_UNREGISTER(to_json);
AST_TEST_UNREGISTER(no_to_ami);
AST_TEST_REGISTER(message_type);
AST_TEST_REGISTER(message);
AST_TEST_REGISTER(subscription_messages);
+ AST_TEST_REGISTER(subscription_pool_messages);
AST_TEST_REGISTER(publish);
AST_TEST_REGISTER(publish_sync);
+ AST_TEST_REGISTER(publish_pool);
AST_TEST_REGISTER(unsubscribe_stops_messages);
AST_TEST_REGISTER(forward);
AST_TEST_REGISTER(cache_filter);
AST_TEST_REGISTER(cache_dump);
AST_TEST_REGISTER(cache_eid_aggregate);
AST_TEST_REGISTER(router);
+ AST_TEST_REGISTER(router_pool);
AST_TEST_REGISTER(router_cache_updates);
AST_TEST_REGISTER(interleaving);
+ AST_TEST_REGISTER(subscription_interleaving);
AST_TEST_REGISTER(no_to_json);
AST_TEST_REGISTER(to_json);
AST_TEST_REGISTER(no_to_ami);
projects / asterisk/asterisk.git / commitdiff