kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-09-03 11:25:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add support for requiring that all queued messages on a caching topic have been handled before

Browse Source 
retrieving from the cache and also change adding channels to an endpoint to be an immediate
operation.

Review: https://reviewboard.asterisk.org/r/2599/


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@391596 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
Joshua Colp
2013-06-13 11:02:16 +00:00
parent 1e9faaf78a
commit 65c492e851
8 changed files with 126 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
main/stasis_cache.c
View File

@@ -46,10 +46,13 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
struct stasis_caching_topic {
struct ao2_container *cache;
struct stasis_topic *topic;
struct stasis_topic *original_topic;
struct stasis_subscription *sub;
snapshot_get_id id_fn;
};
static struct stasis_message_type *cache_guarantee_type(void);
static void stasis_caching_topic_dtor(void *obj) {
struct stasis_caching_topic *caching_topic = obj;
ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
@@ -60,6 +63,8 @@ static void stasis_caching_topic_dtor(void *obj) {
caching_topic->cache = NULL;
ao2_cleanup(caching_topic->topic);
caching_topic->topic = NULL;
ao2_cleanup(caching_topic->original_topic);
caching_topic->original_topic = NULL;
}
struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
@@ -204,13 +209,62 @@ static struct stasis_message *cache_put(struct stasis_caching_topic *caching_top
return old_snapshot;
}
struct stasis_message *stasis_cache_get(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id)
/*! \internal */
struct caching_guarantee {
ast_mutex_t lock;
ast_cond_t cond;
unsigned int done:1;
};
static void caching_guarantee_dtor(void *obj)
{
struct caching_guarantee *guarantee = obj;
ast_assert(guarantee->done == 1);
ast_mutex_destroy(&guarantee->lock);
ast_cond_destroy(&guarantee->cond);
}
static struct stasis_message *caching_guarantee_create(void)
{
RAII_VAR(struct caching_guarantee *, guarantee, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
if (!(guarantee = ao2_alloc(sizeof(*guarantee), caching_guarantee_dtor))) {
return NULL;
}
ast_mutex_init(&guarantee->lock);
ast_cond_init(&guarantee->cond, NULL);
if (!(msg = stasis_message_create(cache_guarantee_type(), guarantee))) {
return NULL;
}
ao2_ref(msg, +1);
return msg;
}
struct stasis_message *stasis_cache_get_extended(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id, unsigned int guaranteed)
{
RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
ast_assert(caching_topic->cache != NULL);
if (guaranteed) {
RAII_VAR(struct stasis_message *, msg, caching_guarantee_create(), ao2_cleanup);
struct caching_guarantee *guarantee = stasis_message_data(msg);
ast_mutex_lock(&guarantee->lock);
stasis_publish(caching_topic->original_topic, msg);
while (!guarantee->done) {
ast_cond_wait(&guarantee->cond, &guarantee->lock);
}
ast_mutex_unlock(&guarantee->lock);
}
search_entry = cache_entry_create(type, id, NULL);
if (search_entry == NULL) {
return NULL;
@@ -261,6 +315,7 @@ struct ao2_container *stasis_cache_dump(struct stasis_caching_topic *caching_top
STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type);
STASIS_MESSAGE_TYPE_DEFN(cache_guarantee_type);
struct stasis_message *stasis_cache_clear_create(struct stasis_message *id_message)
{
@@ -340,6 +395,18 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub, stru
caching_topic_needs_unref = caching_topic;
}
/* Handle cache guarantee event */
if (cache_guarantee_type() == stasis_message_type(message)) {
struct caching_guarantee *guarantee = stasis_message_data(message);
ast_mutex_lock(&guarantee->lock);
guarantee->done = 1;
ast_cond_signal(&guarantee->cond);
ast_mutex_unlock(&guarantee->lock);
return;
}
/* Handle cache clear event */
if (stasis_cache_clear_type() == stasis_message_type(message)) {
RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
@@ -423,6 +490,10 @@ struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *or
if (sub == NULL) {
return NULL;
}
ao2_ref(original_topic, +1);
caching_topic->original_topic = original_topic;
/* This is for the reference contained in the subscription above */
ao2_ref(caching_topic, +1);
caching_topic->sub = sub;
@@ -435,6 +506,7 @@ static void stasis_cache_cleanup(void)
{
STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
STASIS_MESSAGE_TYPE_CLEANUP(cache_guarantee_type);
}
int stasis_cache_init(void)
@@ -449,6 +521,10 @@ int stasis_cache_init(void)
return -1;
}
if (STASIS_MESSAGE_TYPE_INIT(cache_guarantee_type) != 0) {
return -1;
}
return 0;
}