[asterisk-commits] dlee: branch dlee/stasis-core r382486 - in /team/dlee/stasis-core: include/as...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Wed Mar 6 08:43:13 CST 2013
Author: dlee
Date: Wed Mar 6 08:43:09 2013
New Revision: 382486
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=382486
Log:
Backported kmoore's subscription notices, but made my own test for them.
Modified:
team/dlee/stasis-core/include/asterisk/stasis.h
team/dlee/stasis-core/main/channel.c
team/dlee/stasis-core/main/manager.c
team/dlee/stasis-core/main/stasis.c
team/dlee/stasis-core/main/stasis_cache.c
team/dlee/stasis-core/tests/test_stasis.c
Modified: team/dlee/stasis-core/include/asterisk/stasis.h
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/include/asterisk/stasis.h?view=diff&rev=382486&r1=382485&r2=382486
==============================================================================
--- team/dlee/stasis-core/include/asterisk/stasis.h (original)
+++ team/dlee/stasis-core/include/asterisk/stasis.h Wed Mar 6 08:43:09 2013
@@ -130,6 +130,8 @@
* are cleaned up.
*/
+#include "asterisk/utils.h"
+
/*! @{ */
/*!
@@ -261,19 +263,19 @@
/*! @{ */
/*!
+ * \brief Opaque type for a Stasis subscription.
+ * \since 12
+ */
+struct stasis_subscription;
+
+/*!
* \brief Callback function type for Stasis subscriptions.
* \param data Data field provided with subscription.
* \param topic Topic to which the message was published.
* \param message Published message.
* \since 12
*/
-typedef void (*stasis_subscription_cb)(void *data, struct stasis_topic *topic, struct stasis_message *message);
-
-/*!
- * \brief Opaque type for a Stasis subscription.
- * \since 12
- */
-struct stasis_subscription;
+typedef void (*stasis_subscription_cb)(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message);
/*!
* \brief Create a subscription.
@@ -324,6 +326,57 @@
* \since 12
*/
struct stasis_subscription *stasis_forward_all(struct stasis_topic *from_topic, struct stasis_topic *to_topic);
+
+/*!
+ * \brief Get the unique ID for the subscription.
+ *
+ * \param sub Subscription for which to get the unique ID.
+ * \return Unique ID for the subscription.
+ * \since 12
+ */
+const char *stasis_subscription_uniqueid(const struct stasis_subscription *sub);
+
+/*!
+ * \brief Returns whether a subscription is currently subscribed.
+ *
+ * Note that there may still be messages queued up to be dispatched to this
+ * subscription, but the stasis_subscription_final_message() has been enqueued.
+ *
+ * \param sub Subscription to check
+ * \return False (zero) if subscription is not subscribed.
+ * \return True (non-zero) if still subscribed.
+ */
+int stasis_subscription_is_subscribed(const struct stasis_subscription *sub);
+
+/*!
+ * \brief Determine whether a message is the final message to be received on a subscription.
+ *
+ * \param sub Subscription on which the message was received.
+ * \param msg Message to check.
+ * \return zero if the provided message is not the final message.
+ * \return non-zero if the provided message is the final message.
+ * \since 12
+ */
+int stasis_subscription_final_message(struct stasis_subscription *sub, struct stasis_message *msg);
+
+/*!
+ * \brief Holds details about changes to subscriptions for the specified topic
+ * \since 12
+ */
+struct stasis_subscription_change {
+ AST_DECLARE_STRING_FIELDS(
+ AST_STRING_FIELD(uniqueid); /*!< The unique ID associated with this subscription */
+ AST_STRING_FIELD(description); /*!< The description of the change to the subscription associated with the uniqueid */
+ );
+ struct stasis_topic *topic; /*!< The topic the subscription is/was subscribing to */
+};
+
+/*!
+ * \brief Gets the message type for subscription change notices
+ * \return The stasis_message_type for subscription change notices
+ * \since 12
+ */
+struct stasis_message_type *stasis_subscription_change(void);
/*! @} */
Modified: team/dlee/stasis-core/main/channel.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/main/channel.c?view=diff&rev=382486&r1=382485&r2=382486
==============================================================================
--- team/dlee/stasis-core/main/channel.c (original)
+++ team/dlee/stasis-core/main/channel.c Wed Mar 6 08:43:09 2013
@@ -8633,7 +8633,7 @@
__ast_channel_snapshot = NULL;
ao2_cleanup(__ast_channel_events_all);
__ast_channel_events_all = NULL;
- ao2_cleanup(__ast_channel_events_all_cached);
+ stasis_caching_unsubscribe(__ast_channel_events_all_cached);
__ast_channel_events_all_cached = NULL;
ast_data_unregister(NULL);
ast_cli_unregister_multiple(cli_channel, ARRAY_LEN(cli_channel));
Modified: team/dlee/stasis-core/main/manager.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/main/manager.c?view=diff&rev=382486&r1=382485&r2=382486
==============================================================================
--- team/dlee/stasis-core/main/manager.c (original)
+++ team/dlee/stasis-core/main/manager.c Wed Mar 6 08:43:09 2013
@@ -7548,7 +7548,7 @@
channel_name, name, value, uniqueid);
}
-static void channel_event_cb(void *data, struct stasis_topic *topic, struct stasis_message *message)
+static void channel_event_cb(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
{
if (stasis_message_type(message) == stasis_cache_update()) {
struct stasis_cache_update *update = stasis_message_data(message);
Modified: team/dlee/stasis-core/main/stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/main/stasis.c?view=diff&rev=382486&r1=382485&r2=382486
==============================================================================
--- team/dlee/stasis-core/main/stasis.c (original)
+++ team/dlee/stasis-core/main/stasis.c Wed Mar 6 08:43:09 2013
@@ -44,6 +44,8 @@
/*! Threadpool for dispatching notifications to subscribers */
static struct ast_threadpool *pool;
+static struct stasis_message_type *__subscription_change_message_type;
+
/*! \private */
struct stasis_topic {
char *name;
@@ -100,6 +102,8 @@
/*! \private */
struct stasis_subscription {
+ /*! Unique ID for this subscription */
+ char *uniqueid;
/*! Topic subscribed to. */
struct stasis_topic *topic;
/*! Mailbox for processing incoming messages. */
@@ -113,17 +117,22 @@
static void subscription_dtor(void *obj)
{
struct stasis_subscription *sub = obj;
+ ast_assert(!stasis_subscription_is_subscribed(sub));
+ ast_free(sub->uniqueid);
+ sub->uniqueid = NULL;
ao2_cleanup(sub->topic);
sub->topic = NULL;
ao2_cleanup(sub->mailbox);
sub->mailbox = NULL;
}
+static void send_subscription_change_message(struct stasis_topic *topic, char *uniqueid, char *description);
+
static struct stasis_subscription *__stasis_subscribe(struct stasis_topic *topic, stasis_subscription_cb callback, void *data, int needs_mailbox)
{
RAII_VAR(struct stasis_subscription *, sub, NULL, ao2_cleanup);
RAII_VAR(struct ast_uuid *, id, NULL, ast_free);
- char stupid_name[AST_UUID_STR_LEN];
+ char uniqueid[AST_UUID_STR_LEN];
sub = ao2_alloc(sizeof(*sub), subscription_dtor);
if (!sub) {
@@ -135,14 +144,15 @@
ast_log(LOG_ERROR, "UUID generation failed\n");
return NULL;
}
- ast_uuid_to_str(id, stupid_name, sizeof(stupid_name));
+ ast_uuid_to_str(id, uniqueid, sizeof(uniqueid));
if (needs_mailbox) {
- sub->mailbox = ast_threadpool_serializer(stupid_name, pool);
+ sub->mailbox = ast_threadpool_serializer(uniqueid, pool);
if (!sub->mailbox) {
return NULL;
}
}
+ sub->uniqueid = ast_strdup(uniqueid);
ao2_ref(topic, +1);
sub->topic = topic;
sub->callback = callback;
@@ -151,6 +161,7 @@
if (topic_add_subscription(topic, sub) != 0) {
return NULL;
}
+ send_subscription_change_message(topic, uniqueid, "Subscribe");
ao2_ref(sub, +1);
return sub;
@@ -170,6 +181,7 @@
for (i = 0; i < topic->num_subscribers_current; ++i) {
if (topic->subscribers[i] == sub) {
+ send_subscription_change_message(topic, sub->uniqueid, "Unsubscribe");
/* swap [i] with last entry; remove last entry */
topic->subscribers[i] = topic->subscribers[--topic->num_subscribers_current];
/* Unsubscribing unrefs the subscription */
@@ -180,6 +192,47 @@
ast_log(LOG_ERROR, "Internal error: subscription has invalid topic\n");
}
+}
+
+int stasis_subscription_is_subscribed(const struct stasis_subscription *sub)
+{
+ if (sub) {
+ size_t i;
+ struct stasis_topic *topic = sub->topic;
+ SCOPED_AO2LOCK(lock_topic, topic);
+
+ for (i = 0; i < topic->num_subscribers_current; ++i) {
+ if (topic->subscribers[i] == sub) {
+ return 1;
+ }
+ }
+ }
+
+ return 0;
+}
+
+const char *stasis_subscription_uniqueid(const struct stasis_subscription *sub)
+{
+ return sub->uniqueid;
+}
+
+int stasis_subscription_final_message(struct stasis_subscription *sub, struct stasis_message *msg)
+{
+ struct stasis_subscription_change *change;
+ if (stasis_message_type(msg) != stasis_subscription_change()) {
+ return 0;
+ }
+
+ change = stasis_message_data(msg);
+ if (strcmp("Unsubscribe", change->description)) {
+ return 0;
+ }
+
+ if (strcmp(stasis_subscription_uniqueid(sub), change->uniqueid)) {
+ return 0;
+ }
+
+ return 1;
}
/*!
@@ -272,6 +325,7 @@
sub_topic = dispatch->sub->topic;
dispatch->sub->callback(dispatch->sub->data,
+ dispatch->sub,
sub_topic,
dispatch->message);
@@ -320,7 +374,7 @@
}
} else {
/* No mailbox; dispatch directly */
- sub->callback(sub->data, sub->topic, message);
+ sub->callback(sub->data, sub, sub->topic, message);
}
}
@@ -328,7 +382,6 @@
ao2_cleanup(subscribers[i]);
}
ast_free(subscribers);
-
}
void stasis_publish(struct stasis_topic *topic, struct stasis_message *message)
@@ -337,24 +390,86 @@
}
/*! \brief Forwarding subscriber */
-static void stasis_forward_cb(void *data, struct stasis_topic *topic, struct stasis_message *message)
+static void stasis_forward_cb(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
{
struct stasis_topic *to_topic = data;
stasis_forward_message(to_topic, topic, message);
+
+ if (stasis_subscription_final_message(sub, message)) {
+ ao2_cleanup(to_topic);
+ }
}
struct stasis_subscription *stasis_forward_all(struct stasis_topic *from_topic, struct stasis_topic *to_topic)
{
+ struct stasis_subscription *sub;
if (!from_topic || !to_topic) {
return NULL;
}
/* Subscribe without a mailbox, since we're just forwarding messages */
- return __stasis_subscribe(from_topic, stasis_forward_cb, to_topic, 0);
+ sub = __stasis_subscribe(from_topic, stasis_forward_cb, to_topic, 0);
+ if (sub) {
+ /* hold a ref to to_topic for this forwarding subscription */
+ ao2_ref(to_topic, +1);
+ }
+ return sub;
+}
+
+static void subscription_change_dtor(void *obj)
+{
+ struct stasis_subscription_change *change = obj;
+ ast_string_field_free_memory(change);
+ ao2_cleanup(change->topic);
+}
+
+static struct stasis_subscription_change *subscription_change_alloc(struct stasis_topic *topic, char *uniqueid, char *description)
+{
+ RAII_VAR(struct stasis_subscription_change *, change, NULL, ao2_cleanup);
+
+ change = ao2_alloc(sizeof(struct stasis_subscription_change), subscription_change_dtor);
+ if (ast_string_field_init(change, 128)) {
+ return NULL;
+ }
+
+ ast_string_field_set(change, uniqueid, uniqueid);
+ ast_string_field_set(change, description, description);
+ ao2_ref(topic, +1);
+ change->topic = topic;
+
+ ao2_ref(change, +1);
+ return change;
+}
+
+struct stasis_message_type *stasis_subscription_change(void)
+{
+ return __subscription_change_message_type;
+}
+
+static void send_subscription_change_message(struct stasis_topic *topic, char *uniqueid, char *description)
+{
+ RAII_VAR(struct stasis_subscription_change *, change, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+
+ change = subscription_change_alloc(topic, uniqueid, description);
+
+ if (!change) {
+ return;
+ }
+
+ msg = stasis_message_create(stasis_subscription_change(), change);
+
+ if (!msg) {
+ return;
+ }
+
+ stasis_publish(topic, msg);
}
/*! \brief Cleanup function */
static void stasis_exit(void)
{
+ ao2_cleanup(__subscription_change_message_type);
+ __subscription_change_message_type = NULL;
ast_threadpool_shutdown(pool);
pool = NULL;
}
@@ -390,5 +505,10 @@
return -1;
}
+ __subscription_change_message_type = stasis_message_type_create("stasis_subscription_change");
+ if (!__subscription_change_message_type) {
+ return -1;
+ }
+
return 0;
}
Modified: team/dlee/stasis-core/main/stasis_cache.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/main/stasis_cache.c?view=diff&rev=382486&r1=382485&r2=382486
==============================================================================
--- team/dlee/stasis-core/main/stasis_cache.c (original)
+++ team/dlee/stasis-core/main/stasis_cache.c Wed Mar 6 08:43:09 2013
@@ -68,7 +68,6 @@
if (caching_topic->sub) {
stasis_unsubscribe(caching_topic->sub);
caching_topic->sub = NULL;
- ao2_cleanup(caching_topic);
} else {
ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
}
@@ -313,7 +312,7 @@
return msg;
}
-static void caching_topic_exec(void *data, struct stasis_topic *topic, struct stasis_message *message)
+static void caching_topic_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
{
struct stasis_caching_topic *caching_topic = data;
const char *id = NULL;
@@ -358,6 +357,10 @@
stasis_publish(caching_topic->topic, update);
}
+
+ if (stasis_subscription_final_message(sub, message)) {
+ ao2_cleanup(caching_topic);
+ }
}
struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, snapshot_get_id id_fn)
Modified: team/dlee/stasis-core/tests/test_stasis.c
URL: http://svnview.digium.com/svn/asterisk/team/dlee/stasis-core/tests/test_stasis.c?view=diff&rev=382486&r1=382485&r2=382486
==============================================================================
--- team/dlee/stasis-core/tests/test_stasis.c (original)
+++ team/dlee/stasis-core/tests/test_stasis.c Wed Mar 6 08:43:09 2013
@@ -115,6 +115,8 @@
ast_cond_t out;
struct stasis_message **messages_rxed;
size_t messages_rxed_len;
+ int ignore_subscriptions;
+ int complete;
};
static void consumer_dtor(void *obj) {
@@ -127,9 +129,10 @@
ao2_cleanup(consumer->messages_rxed[--consumer->messages_rxed_len]);
}
ast_free(consumer->messages_rxed);
-}
-
-static struct consumer *consumer_create(void) {
+ consumer->messages_rxed = NULL;
+}
+
+static struct consumer *consumer_create(int ignore_subscriptions) {
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
@@ -138,7 +141,11 @@
return NULL;
}
+ consumer->ignore_subscriptions = ignore_subscriptions;
consumer->messages_rxed = ast_malloc(0);
+ if (!consumer->messages_rxed) {
+ return NULL;
+ }
ast_mutex_init(&consumer->lock);
ast_cond_init(&consumer->out, NULL);
@@ -147,23 +154,35 @@
return consumer;
}
-static void consumer_exec(void *data, struct stasis_topic *topic, struct stasis_message *message) {
+static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message)
+{
struct consumer *consumer = data;
- SCOPED_MUTEX(lock, &consumer->lock);
- ++consumer->messages_rxed_len;
- consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
- ast_assert(consumer->messages_rxed != NULL);
- consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
- ao2_ref(message, +1);
-
- ast_cond_signal(&consumer->out);
-}
-
-static int consumer_wait_for(struct consumer *consumer, size_t expected_len) {
+ if (!consumer->ignore_subscriptions || stasis_message_type(message) != stasis_subscription_change()) {
+ SCOPED_MUTEX(lock, &consumer->lock);
+
+ ++consumer->messages_rxed_len;
+ consumer->messages_rxed = ast_realloc(consumer->messages_rxed, sizeof(*consumer->messages_rxed) * consumer->messages_rxed_len);
+ ast_assert(consumer->messages_rxed != NULL);
+ consumer->messages_rxed[consumer->messages_rxed_len - 1] = message;
+ ao2_ref(message, +1);
+
+ ast_cond_signal(&consumer->out);
+ }
+
+ if (stasis_subscription_final_message(sub, message)) {
+ SCOPED_MUTEX(lock, &consumer->lock);
+ consumer->complete = 1;
+ ast_cond_signal(&consumer->out);
+ ao2_cleanup(consumer);
+ }
+}
+
+static int consumer_wait_for(struct consumer *consumer, size_t expected_len)
+{
struct timeval start = ast_tvnow();
struct timespec end = {
- .tv_sec = start.tv_sec + 3600,
+ .tv_sec = start.tv_sec + 30,
.tv_nsec = start.tv_usec * 1000
};
@@ -177,6 +196,26 @@
ast_assert(r == 0); /* Not expecting any othet types of errors */
}
return consumer->messages_rxed_len;
+}
+
+static int consumer_wait_for_completion(struct consumer *consumer)
+{
+ struct timeval start = ast_tvnow();
+ struct timespec end = {
+ .tv_sec = start.tv_sec + 30,
+ .tv_nsec = start.tv_usec * 1000
+ };
+
+ SCOPED_MUTEX(lock, &consumer->lock);
+
+ while (!consumer->complete) {
+ int r = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
+ if (r == ETIMEDOUT) {
+ break;
+ }
+ ast_assert(r == 0); /* Not expecting any othet types of errors */
+ }
+ return consumer->complete;
}
static int consumer_should_stay(struct consumer *consumer, size_t expected_len) {
@@ -203,7 +242,64 @@
return consumer->messages_rxed_len;
}
-AST_TEST_DEFINE(subscribe)
+AST_TEST_DEFINE(subscription_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";
+ info->description = "Test subscribe/unsubscribe messages";
+ 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(topic, consumer_exec, consumer);
+ ast_test_validate(test, NULL != uut);
+ ao2_ref(consumer, +1);
+ expected_uniqueid = ast_strdup(stasis_subscription_uniqueid(uut));
+
+ stasis_unsubscribe(uut);
+ uut = NULL;
+ complete = consumer_wait_for_completion(consumer);
+ ast_test_validate(test, 1 == complete);
+
+ printf("%zd\n", consumer->messages_rxed_len);
+ ast_test_validate(test, 2 == consumer->messages_rxed_len);
+ ast_test_validate(test, stasis_subscription_change() == stasis_message_type(consumer->messages_rxed[0]));
+ ast_test_validate(test, stasis_subscription_change() == 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);
RAII_VAR(struct stasis_subscription *, uut, NULL, stasis_unsubscribe);
@@ -228,11 +324,12 @@
topic = stasis_topic_create("TestTopic");
ast_test_validate(test, NULL != topic);
- consumer = consumer_create();
+ consumer = consumer_create(1);
ast_test_validate(test, NULL != consumer);
uut = stasis_subscribe(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);
@@ -249,7 +346,7 @@
return AST_TEST_PASS;
}
-AST_TEST_DEFINE(unsubscribe)
+AST_TEST_DEFINE(unsubscribe_stops_messages)
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
@@ -273,11 +370,12 @@
topic = stasis_topic_create("TestTopic");
ast_test_validate(test, NULL != topic);
- consumer = consumer_create();
+ consumer = consumer_create(1);
ast_test_validate(test, NULL != consumer);
uut = stasis_subscribe(topic, consumer_exec, consumer);
ast_test_validate(test, NULL != uut);
+ ao2_ref(consumer, +1);
stasis_unsubscribe(uut);
uut = NULL;
@@ -335,15 +433,17 @@
forward_sub = stasis_forward_all(topic, parent_topic);
ast_test_validate(test, NULL != forward_sub);
- parent_consumer = consumer_create();
+ parent_consumer = consumer_create(1);
ast_test_validate(test, NULL != parent_consumer);
- consumer = consumer_create();
+ consumer = consumer_create(1);
ast_test_validate(test, NULL != consumer);
parent_sub = stasis_subscribe(parent_topic, consumer_exec, parent_consumer);
ast_test_validate(test, NULL != parent_sub);
+ ao2_ref(parent_consumer, +1);
sub = stasis_subscribe(topic, consumer_exec, consumer);
ast_test_validate(test, NULL != sub);
+ ao2_ref(consumer, +1);
test_data = ao2_alloc(1, NULL);
ast_test_validate(test, NULL != test_data);
@@ -430,10 +530,11 @@
ast_test_validate(test, NULL != topic);
caching_topic = stasis_caching_topic_create(topic, cache_test_data_id);
ast_test_validate(test, NULL != caching_topic);
- consumer = consumer_create();
+ consumer = consumer_create(1);
ast_test_validate(test, NULL != consumer);
sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
ast_test_validate(test, NULL != sub);
+ ao2_ref(consumer, +1);
test_message = cache_test_message_create(non_cache_type, "1", "1");
ast_test_validate(test, NULL != test_message);
@@ -483,10 +584,11 @@
ast_test_validate(test, NULL != topic);
caching_topic = stasis_caching_topic_create(topic, cache_test_data_id);
ast_test_validate(test, NULL != caching_topic);
- consumer = consumer_create();
+ consumer = consumer_create(1);
ast_test_validate(test, NULL != consumer);
sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, consumer);
ast_test_validate(test, NULL != sub);
+ ao2_ref(consumer, +1);
test_message1_1 = cache_test_message_create(cache_type, "1", "1");
ast_test_validate(test, NULL != test_message1_1);
@@ -550,8 +652,9 @@
{
AST_TEST_UNREGISTER(message_type);
AST_TEST_UNREGISTER(message);
- AST_TEST_UNREGISTER(subscribe);
- AST_TEST_UNREGISTER(unsubscribe);
+ AST_TEST_UNREGISTER(subscription_messages);
+ AST_TEST_UNREGISTER(publish);
+ AST_TEST_UNREGISTER(unsubscribe_stops_messages);
AST_TEST_UNREGISTER(forward);
AST_TEST_UNREGISTER(cache_passthrough);
AST_TEST_UNREGISTER(cache);
@@ -562,8 +665,9 @@
{
AST_TEST_REGISTER(message_type);
AST_TEST_REGISTER(message);
- AST_TEST_REGISTER(subscribe);
- AST_TEST_REGISTER(unsubscribe);
+ AST_TEST_REGISTER(subscription_messages);
+ AST_TEST_REGISTER(publish);
+ AST_TEST_REGISTER(unsubscribe_stops_messages);
AST_TEST_REGISTER(forward);
AST_TEST_REGISTER(cache_passthrough);
AST_TEST_REGISTER(cache);
More information about the asterisk-commits
mailing list