[asterisk-commits] dlee: trunk r383242 - in /trunk: include/asterisk/ main/ tests/
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Fri Mar 15 12:35:20 CDT 2013
Author: dlee
Date: Fri Mar 15 12:35:16 2013
New Revision: 383242
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=383242
Log:
A simplistic router for stasis_message's.
Often times, when subscribing to a topic, one wants to handle
different message types differently. While one could cascade if/else
statements through the subscription handler, it is much cleaner to
specify a different callback for each message type. The
stasis_message_router is here to help!
A stasis_message_router is constructed for a particular stasis_topic,
which is subscribes to. Call stasis_message_router_unsubscribe() to
cancel that subscription.
Once constructed, routes can be added using
stasis_message_router_add() (or stasis_message_router_set_default()
for any messages not handled by other routes). There may be only one
route per stasis_message_type. The route's callback is invoked just as
if it were a callback for a subscription; but it only gets called for
messages of the specified type.
(issue ASTERISK-20887)
Added:
trunk/include/asterisk/stasis_message_router.h
- copied unchanged from r383241, team/dlee/stasis-router/include/asterisk/stasis_message_router.h
trunk/main/stasis_message_router.c
- copied unchanged from r383241, team/dlee/stasis-router/main/stasis_message_router.c
Modified:
trunk/main/stasis.c
trunk/main/stasis_cache.c
trunk/main/stasis_message.c
trunk/tests/test_stasis.c
Modified: trunk/main/stasis.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/stasis.c?view=diff&rev=383242&r1=383241&r2=383242
==============================================================================
--- trunk/main/stasis.c (original)
+++ trunk/main/stasis.c Fri Mar 15 12:35:16 2013
@@ -46,7 +46,7 @@
static struct stasis_message_type *__subscription_change_message_type;
-/*! \private */
+/*! \internal */
struct stasis_topic {
char *name;
/*! Variable length array of the subscribers (raw pointer to avoid cyclic references) */
@@ -86,7 +86,7 @@
}
topic->num_subscribers_max = INITIAL_SUBSCRIBERS_MAX;
- topic->subscribers = ast_calloc(topic->num_subscribers_max, sizeof(topic->subscribers));
+ topic->subscribers = ast_calloc(topic->num_subscribers_max, sizeof(*topic->subscribers));
if (!topic->subscribers) {
return NULL;
}
@@ -100,7 +100,7 @@
return topic->name;
}
-/*! \private */
+/*! \internal */
struct stasis_subscription {
/*! Unique ID for this subscription */
char *uniqueid;
@@ -264,7 +264,7 @@
}
/*!
- * \private
+ * \internal
* \brief Information needed to dispatch a message to a subscription
*/
struct dispatch {
Modified: trunk/main/stasis_cache.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/stasis_cache.c?view=diff&rev=383242&r1=383241&r2=383242
==============================================================================
--- trunk/main/stasis_cache.c (original)
+++ trunk/main/stasis_cache.c Fri Mar 15 12:35:16 2013
@@ -42,7 +42,7 @@
#define NUM_CACHE_BUCKETS 563
#endif
-/*! \private */
+/*! \internal */
struct stasis_caching_topic {
struct ao2_container *cache;
struct stasis_topic *topic;
Modified: trunk/main/stasis_message.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/stasis_message.c?view=diff&rev=383242&r1=383241&r2=383242
==============================================================================
--- trunk/main/stasis_message.c (original)
+++ trunk/main/stasis_message.c Fri Mar 15 12:35:16 2013
@@ -35,7 +35,7 @@
#include "asterisk/stasis.h"
#include "asterisk/utils.h"
-/*! \private */
+/*! \internal */
struct stasis_message_type {
char *name;
};
@@ -70,7 +70,7 @@
return type->name;
}
-/*! \private */
+/*! \internal */
struct stasis_message {
/*! Time the message was created */
struct timeval timestamp;
Modified: trunk/tests/test_stasis.c
URL: http://svnview.digium.com/svn/asterisk/trunk/tests/test_stasis.c?view=diff&rev=383242&r1=383241&r2=383242
==============================================================================
--- trunk/tests/test_stasis.c (original)
+++ trunk/tests/test_stasis.c Fri Mar 15 12:35:16 2013
@@ -36,6 +36,7 @@
#include "asterisk/astobj2.h"
#include "asterisk/module.h"
#include "asterisk/stasis.h"
+#include "asterisk/stasis_message_router.h"
#include "asterisk/test.h"
static const char *test_category = "/stasis/core/";
@@ -674,6 +675,148 @@
return AST_TEST_PASS;
}
+AST_TEST_DEFINE(route_conflicts)
+{
+ RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe);
+ RAII_VAR(struct stasis_message_type *, test_message_type, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer1, NULL, ao2_cleanup);
+ RAII_VAR(struct consumer *, consumer2, NULL, ao2_cleanup);
+ int ret;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = __func__;
+ info->category = test_category;
+ info->summary =
+ "Multiple routes to the same message_type should fail";
+ info->description =
+ "Multiple routes to the same message_type should fail";
+ 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);
+
+ test_message_type = stasis_message_type_create("TestMessage");
+ ast_test_validate(test, NULL != test_message_type);
+
+ uut = stasis_message_router_create(topic);
+ ast_test_validate(test, NULL != uut);
+
+ ret = stasis_message_router_add(
+ uut, test_message_type, consumer_exec, consumer1);
+ ast_test_validate(test, 0 == ret);
+ ret = stasis_message_router_add(
+ uut, test_message_type, consumer_exec, consumer2);
+ ast_test_validate(test, 0 != ret);
+
+ return AST_TEST_PASS;
+}
+
+AST_TEST_DEFINE(router)
+{
+ RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
+ RAII_VAR(struct stasis_message_router *, uut, NULL, stasis_message_router_unsubscribe);
+ 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 simple message routing";
+ info->description = "Test simple message routing";
+ 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");
+ ast_test_validate(test, NULL != test_message_type1);
+ test_message_type2 = stasis_message_type_create("TestMessage2");
+ ast_test_validate(test, NULL != test_message_type2);
+ test_message_type3 = stasis_message_type_create("TestMessage3");
+ ast_test_validate(test, NULL != test_message_type3);
+
+ uut = stasis_message_router_create(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 int unload_module(void)
{
AST_TEST_UNREGISTER(message_type);
@@ -684,6 +827,8 @@
AST_TEST_UNREGISTER(forward);
AST_TEST_UNREGISTER(cache_passthrough);
AST_TEST_UNREGISTER(cache);
+ AST_TEST_UNREGISTER(route_conflicts);
+ AST_TEST_UNREGISTER(router);
return 0;
}
@@ -697,6 +842,8 @@
AST_TEST_REGISTER(forward);
AST_TEST_REGISTER(cache_passthrough);
AST_TEST_REGISTER(cache);
+ AST_TEST_REGISTER(route_conflicts);
+ AST_TEST_REGISTER(router);
return AST_MODULE_LOAD_SUCCESS;
}
More information about the asterisk-commits
mailing list