kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-10-24 21:50:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Tweak caching topics to fix CEL tests

Browse Source 
The Stasis changes in r395954 had an unanticipated side effect: messages
published directly to an _all topic does not get forwarded to the
corresponding caching topic.

This patch fixes that by changing how caching topics forward messages,
and how the caching pattern forwards are setup.

For the caching pattern, the all_topic is forwarded to the
all_topic_cached. This forwards messages published directly to the
all_topic to all_topic_cached.

In order to avoid duplicate messages on all_topic_cached, caching topics
were changed to no longer forward uncached messages. Subscribers to an
individual caching topic should only expect to receive cache updates,
and subscription change messages. Since individual caching topics are
new, this shouldn't be a problem.

There are a few minor changes to the pre-cache split behavior.

 * For topics changed to use the caching pattern, the all_topic_cached
   will forward snapshots in addition to cache updates. Since
   subscribers by design ignore unexpected messages, this should be
   fine.

 * Caching topics that don't use the caching pattern no longer forward
   non-cache updates. This makes no difference for the current caching
   topics.

   * mwi_topic_cached, channel_by_name_topic and
     presence_state_topic_cached have no subscribers

   * device_state_topic_cached's only subscriber only processes cache
     udpates

(issue ASTERISK-22243)
Review: https://reviewboard.asterisk.org/r/2738


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@396329 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
David M. Lee
2013-08-06 14:28:23 +00:00
parent dfa5be76d4
commit b97d318b7b
4 changed files with 40 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
include/asterisk/stasis.h
View File

@@ -110,12 +110,12 @@
* It's a thread safe container, so freely use the stasis_cache_get() and * It's a thread safe container, so freely use the stasis_cache_get() and
* stasis_cache_dump() to query the cache. * stasis_cache_dump() to query the cache.
* *
* The \ref stasis_caching_topic provides a topic that forwards non-cacheable * The \ref stasis_caching_topic discards non-cacheable messages. A cacheable
* messages unchanged. A cacheable message is wrapped in a \ref * message is wrapped in a \ref stasis_cache_update message which provides the
* stasis_cache_update message which provides the old snapshot (or \c NULL if * old snapshot (or \c NULL if this is a new cache entry), and the new snapshot
* this is a new cache entry), and the new snapshot (or \c NULL if the entry was * (or \c NULL if the entry was removed from the cache). A
* removed from the cache). A stasis_cache_clear_create() message must be sent * stasis_cache_clear_create() message must be sent to the topic in order to
* to the topic in order to remove entries from the cache. * remove entries from the cache.
* *
* In order to unsubscribe a \ref stasis_caching_topic from the upstream topic, * In order to unsubscribe a \ref stasis_caching_topic from the upstream topic,
* call stasis_caching_unsubscribe(). Due to cyclic references, the \ref * call stasis_caching_unsubscribe(). Due to cyclic references, the \ref

3
main/stasis_cache.c
View File

@@ -426,8 +426,7 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub,
id = caching_topic->cache->id_fn(message); id = caching_topic->cache->id_fn(message);
if (id == NULL) { if (id == NULL) {
/* Object isn't cached; forward */ /* Object isn't cached; discard */
stasis_forward_message(caching_topic->topic, topic, message);
} else { } else {
/* Update the cache */ /* Update the cache */
RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);

37
main/stasis_cache_pattern.c
View File

@@ -38,14 +38,16 @@ struct stasis_cp_all {
struct stasis_topic *topic; struct stasis_topic *topic;
struct stasis_topic *topic_cached; struct stasis_topic *topic_cached;
struct stasis_cache *cache; struct stasis_cache *cache;
struct stasis_subscription *forward_all_to_cached;
}; };
struct stasis_cp_single { struct stasis_cp_single {
struct stasis_topic *topic; struct stasis_topic *topic;
struct stasis_caching_topic *topic_cached; struct stasis_caching_topic *topic_cached;
struct stasis_subscription *forward; struct stasis_subscription *forward_topic_to_all;
struct stasis_subscription *forward_cached; struct stasis_subscription *forward_cached_to_all;
}; };
static void all_dtor(void *obj) static void all_dtor(void *obj)
@@ -53,8 +55,13 @@ static void all_dtor(void *obj)
struct stasis_cp_all *all = obj; struct stasis_cp_all *all = obj;
ao2_cleanup(all->topic); ao2_cleanup(all->topic);
all->topic = NULL;
ao2_cleanup(all->topic_cached); ao2_cleanup(all->topic_cached);
all->topic_cached = NULL;
ao2_cleanup(all->cache); ao2_cleanup(all->cache);
all->cache = NULL;
stasis_unsubscribe_and_join(all->forward_all_to_cached);
all->forward_all_to_cached = NULL;
} }
struct stasis_cp_all *stasis_cp_all_create(const char *name, struct stasis_cp_all *stasis_cp_all_create(const char *name,
@@ -76,8 +83,11 @@ struct stasis_cp_all *stasis_cp_all_create(const char *name,
all->topic = stasis_topic_create(name); all->topic = stasis_topic_create(name);
all->topic_cached = stasis_topic_create(cached_name); all->topic_cached = stasis_topic_create(cached_name);
all->cache = stasis_cache_create(id_fn); all->cache = stasis_cache_create(id_fn);
all->forward_all_to_cached =
stasis_forward_all(all->topic, all->topic_cached);
if (!all->topic || !all->topic_cached || !all->cache) { if (!all->topic || !all->topic_cached || !all->cache ||
!all->forward_all_to_cached) {
return NULL; return NULL;
} }
@@ -116,8 +126,8 @@ static void one_dtor(void *obj)
/* Should already be unsubscribed */ /* Should already be unsubscribed */
ast_assert(one->topic_cached == NULL); ast_assert(one->topic_cached == NULL);
ast_assert(one->forward == NULL); ast_assert(one->forward_topic_to_all == NULL);
ast_assert(one->forward_cached == NULL); ast_assert(one->forward_cached_to_all == NULL);
ao2_cleanup(one->topic); ao2_cleanup(one->topic);
one->topic = NULL; one->topic = NULL;
@@ -142,13 +152,13 @@ struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all,
return NULL; return NULL;
} }
one->forward = stasis_forward_all(one->topic, all->topic); one->forward_topic_to_all = stasis_forward_all(one->topic, all->topic);
if (!one->forward) { if (!one->forward_topic_to_all) {
return NULL; return NULL;
} }
one->forward_cached = stasis_forward_all( one->forward_cached_to_all = stasis_forward_all(
stasis_caching_get_topic(one->topic_cached), all->topic_cached); stasis_caching_get_topic(one->topic_cached), all->topic_cached);
if (!one->forward_cached) { if (!one->forward_cached_to_all) {
return NULL; return NULL;
} }
@@ -162,12 +172,11 @@ void stasis_cp_single_unsubscribe(struct stasis_cp_single *one)
return; return;
} }
stasis_unsubscribe(one->forward_topic_to_all);
one->forward_topic_to_all = NULL;
stasis_unsubscribe(one->forward_cached_to_all);
one->forward_cached_to_all = NULL;
stasis_caching_unsubscribe(one->topic_cached); stasis_caching_unsubscribe(one->topic_cached);
one->topic_cached = NULL;
stasis_unsubscribe(one->forward);
one->forward = NULL;
stasis_unsubscribe(one->forward_cached);
one->forward_cached = NULL;
} }
struct stasis_topic *stasis_cp_single_topic(struct stasis_cp_single *one) struct stasis_topic *stasis_cp_single_topic(struct stasis_cp_single *one)

28
tests/test_stasis.c
View File

@@ -610,7 +610,7 @@ static const char *cache_test_data_id(struct stasis_message *message) {
return cachable->id; return cachable->id;
} }
AST_TEST_DEFINE(cache_passthrough) AST_TEST_DEFINE(cache_filter)
{ {
RAII_VAR(struct stasis_message_type *, non_cache_type, NULL, ao2_cleanup); RAII_VAR(struct stasis_message_type *, non_cache_type, NULL, ao2_cleanup);
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup); RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
@@ -620,14 +620,13 @@ AST_TEST_DEFINE(cache_passthrough)
RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe); RAII_VAR(struct stasis_subscription *, sub, NULL, stasis_unsubscribe);
RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, test_message, NULL, ao2_cleanup);
int actual_len; int actual_len;
struct stasis_message_type *actual_type;
switch (cmd) { switch (cmd) {
case TEST_INIT: case TEST_INIT:
info->name = __func__; info->name = __func__;
info->category = test_category; info->category = test_category;
info->summary = "Test passing messages through cache topic unscathed."; info->summary = "Test caching topics only forward cache_update messages.";
info->description = "Test passing messages through cache topic unscathed."; info->description = "Test caching topics only forward cache_update messages.";
return AST_TEST_NOT_RUN; return AST_TEST_NOT_RUN;
case TEST_EXECUTE: case TEST_EXECUTE:
break; break;
@@ -652,13 +651,8 @@ AST_TEST_DEFINE(cache_passthrough)
stasis_publish(topic, test_message); stasis_publish(topic, test_message);
actual_len = consumer_wait_for(consumer, 1); actual_len = consumer_should_stay(consumer, 0);
ast_test_validate(test, 1 == actual_len); ast_test_validate(test, 0 == actual_len);
actual_type = stasis_message_type(consumer->messages_rxed[0]);
ast_test_validate(test, non_cache_type == actual_type);
ast_test_validate(test, test_message == consumer->messages_rxed[0]);
return AST_TEST_PASS; return AST_TEST_PASS;
} }
@@ -1113,8 +1107,9 @@ AST_TEST_DEFINE(router_cache_updates)
ast_test_validate(test, 1 == actual_len); ast_test_validate(test, 1 == actual_len);
actual_len = consumer_wait_for(consumer2, 1); actual_len = consumer_wait_for(consumer2, 1);
ast_test_validate(test, 1 == actual_len); ast_test_validate(test, 1 == actual_len);
actual_len = consumer_wait_for(consumer3, 1); /* Uncacheable message should not be passed through */
ast_test_validate(test, 1 == actual_len); actual_len = consumer_should_stay(consumer3, 0);
ast_test_validate(test, 0 == actual_len);
actual = consumer1->messages_rxed[0]; actual = consumer1->messages_rxed[0];
ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual)); ast_test_validate(test, stasis_cache_update_type() == stasis_message_type(actual));
@@ -1128,9 +1123,6 @@ AST_TEST_DEFINE(router_cache_updates)
ast_test_validate(test, test_message_type2 == update->type); ast_test_validate(test, test_message_type2 == update->type);
ast_test_validate(test, test_message2 == update->new_snapshot); ast_test_validate(test, test_message2 == update->new_snapshot);
actual = consumer3->messages_rxed[0];
ast_test_validate(test, test_message3 == actual);
/* consumer1 and consumer2 do not get the final message. */ /* consumer1 and consumer2 do not get the final message. */
ao2_cleanup(consumer1); ao2_cleanup(consumer1);
ao2_cleanup(consumer2); ao2_cleanup(consumer2);
@@ -1287,7 +1279,7 @@ static int unload_module(void)
AST_TEST_UNREGISTER(publish); AST_TEST_UNREGISTER(publish);
AST_TEST_UNREGISTER(unsubscribe_stops_messages); AST_TEST_UNREGISTER(unsubscribe_stops_messages);
AST_TEST_UNREGISTER(forward); AST_TEST_UNREGISTER(forward);
AST_TEST_UNREGISTER(cache_passthrough); AST_TEST_UNREGISTER(cache_filter);
AST_TEST_UNREGISTER(cache); AST_TEST_UNREGISTER(cache);
AST_TEST_UNREGISTER(cache_dump); AST_TEST_UNREGISTER(cache_dump);
AST_TEST_UNREGISTER(route_conflicts); AST_TEST_UNREGISTER(route_conflicts);
@@ -1309,7 +1301,7 @@ static int load_module(void)
AST_TEST_REGISTER(publish); AST_TEST_REGISTER(publish);
AST_TEST_REGISTER(unsubscribe_stops_messages); AST_TEST_REGISTER(unsubscribe_stops_messages);
AST_TEST_REGISTER(forward); AST_TEST_REGISTER(forward);
AST_TEST_REGISTER(cache_passthrough); AST_TEST_REGISTER(cache_filter);
AST_TEST_REGISTER(cache); AST_TEST_REGISTER(cache);
AST_TEST_REGISTER(cache_dump); AST_TEST_REGISTER(cache_dump);
AST_TEST_REGISTER(route_conflicts); AST_TEST_REGISTER(route_conflicts);