diff options
author | David M. Lee <dlee@digium.com> | 2013-08-06 14:28:23 +0000 |
---|---|---|
committer | David M. Lee <dlee@digium.com> | 2013-08-06 14:28:23 +0000 |
commit | b97d318b7bc31b47fbd4b74421e351095f05139d (patch) | |
tree | 88f67aaab10f6a9322ceef36b285e4caf0a9f2d6 /main | |
parent | dfa5be76d4891f5240440cb5941374893a9ff57b (diff) |
Tweak caching topics to fix CEL tests
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
Diffstat (limited to 'main')
-rw-r--r-- | main/stasis_cache.c | 3 | ||||
-rw-r--r-- | main/stasis_cache_pattern.c | 37 |
2 files changed, 24 insertions, 16 deletions
diff --git a/main/stasis_cache.c b/main/stasis_cache.c index 3d5065665..2ca4083df 100644 --- a/main/stasis_cache.c +++ b/main/stasis_cache.c @@ -426,8 +426,7 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub, id = caching_topic->cache->id_fn(message); if (id == NULL) { - /* Object isn't cached; forward */ - stasis_forward_message(caching_topic->topic, topic, message); + /* Object isn't cached; discard */ } else { /* Update the cache */ RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup); diff --git a/main/stasis_cache_pattern.c b/main/stasis_cache_pattern.c index 18ae8617e..0ee57ba24 100644 --- a/main/stasis_cache_pattern.c +++ b/main/stasis_cache_pattern.c @@ -38,14 +38,16 @@ struct stasis_cp_all { struct stasis_topic *topic; struct stasis_topic *topic_cached; struct stasis_cache *cache; + + struct stasis_subscription *forward_all_to_cached; }; struct stasis_cp_single { struct stasis_topic *topic; struct stasis_caching_topic *topic_cached; - struct stasis_subscription *forward; - struct stasis_subscription *forward_cached; + struct stasis_subscription *forward_topic_to_all; + struct stasis_subscription *forward_cached_to_all; }; static void all_dtor(void *obj) @@ -53,8 +55,13 @@ static void all_dtor(void *obj) struct stasis_cp_all *all = obj; ao2_cleanup(all->topic); + all->topic = NULL; ao2_cleanup(all->topic_cached); + all->topic_cached = NULL; 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, @@ -76,8 +83,11 @@ struct stasis_cp_all *stasis_cp_all_create(const char *name, all->topic = stasis_topic_create(name); all->topic_cached = stasis_topic_create(cached_name); 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; } @@ -116,8 +126,8 @@ static void one_dtor(void *obj) /* Should already be unsubscribed */ ast_assert(one->topic_cached == NULL); - ast_assert(one->forward == NULL); - ast_assert(one->forward_cached == NULL); + ast_assert(one->forward_topic_to_all == NULL); + ast_assert(one->forward_cached_to_all == NULL); ao2_cleanup(one->topic); one->topic = NULL; @@ -142,13 +152,13 @@ struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all, return NULL; } - one->forward = stasis_forward_all(one->topic, all->topic); - if (!one->forward) { + one->forward_topic_to_all = stasis_forward_all(one->topic, all->topic); + if (!one->forward_topic_to_all) { 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); - if (!one->forward_cached) { + if (!one->forward_cached_to_all) { return NULL; } @@ -162,12 +172,11 @@ void stasis_cp_single_unsubscribe(struct stasis_cp_single *one) 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); - 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) |