summaryrefslogtreecommitdiff
path: root/main/stasis_cache.c
diff options
context:
space:
mode:
Diffstat (limited to 'main/stasis_cache.c')
-rw-r--r--main/stasis_cache.c650
1 files changed, 524 insertions, 126 deletions
diff --git a/main/stasis_cache.c b/main/stasis_cache.c
index 1c5053915..22ace60af 100644
--- a/main/stasis_cache.c
+++ b/main/stasis_cache.c
@@ -36,6 +36,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/stasis_internal.h"
#include "asterisk/stasis.h"
#include "asterisk/utils.h"
+#include "asterisk/vector.h"
#ifdef LOW_MEMORY
#define NUM_CACHE_BUCKETS 17
@@ -47,6 +48,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
struct stasis_cache {
struct ao2_container *entries;
snapshot_get_id id_fn;
+ cache_aggregate_calc_fn aggregate_calc_fn;
+ cache_aggregate_publish_fn aggregate_publish_fn;
};
/*! \internal */
@@ -124,29 +127,53 @@ struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_c
return NULL;
}
-struct cache_entry {
+struct cache_entry_key {
struct stasis_message_type *type;
- char *id;
- struct stasis_message *snapshot;
+ const char *id;
+};
+
+struct stasis_cache_entry {
+ struct cache_entry_key key;
+ /*! Aggregate snapshot of the stasis cache. */
+ struct stasis_message *aggregate;
+ /*! Local entity snapshot of the stasis event. */
+ struct stasis_message *local;
+ /*! Remote entity snapshots of the stasis event. */
+ AST_VECTOR(, struct stasis_message *) remote;
};
static void cache_entry_dtor(void *obj)
{
- struct cache_entry *entry = obj;
- ao2_cleanup(entry->type);
- entry->type = NULL;
- ast_free(entry->id);
- entry->id = NULL;
- ao2_cleanup(entry->snapshot);
- entry->snapshot = NULL;
+ struct stasis_cache_entry *entry = obj;
+ size_t idx;
+
+ ao2_cleanup(entry->key.type);
+ entry->key.type = NULL;
+ ast_free((char *) entry->key.id);
+ entry->key.id = NULL;
+
+ ao2_cleanup(entry->aggregate);
+ entry->aggregate = NULL;
+ ao2_cleanup(entry->local);
+ entry->local = NULL;
+
+ for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
+ struct stasis_message *remote;
+
+ remote = AST_VECTOR_GET(&entry->remote, idx);
+ ao2_cleanup(remote);
+ }
+ AST_VECTOR_FREE(&entry->remote);
}
-static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
+static struct stasis_cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
{
- RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
+ struct stasis_cache_entry *entry;
+ int is_remote;
ast_assert(type != NULL);
ast_assert(id != NULL);
+ ast_assert(snapshot != NULL);
entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
AO2_ALLOC_OPT_LOCK_NOLOCK);
@@ -154,173 +181,528 @@ static struct cache_entry *cache_entry_create(struct stasis_message_type *type,
return NULL;
}
- entry->id = ast_strdup(id);
- if (!entry->id) {
+ entry->key.id = ast_strdup(id);
+ if (!entry->key.id) {
+ ao2_cleanup(entry);
+ return NULL;
+ }
+ entry->key.type = ao2_bump(type);
+
+ is_remote = ast_eid_cmp(&ast_eid_default, stasis_message_eid(snapshot)) ? 1 : 0;
+ if (AST_VECTOR_INIT(&entry->remote, is_remote)) {
+ ao2_cleanup(entry);
return NULL;
}
- ao2_ref(type, +1);
- entry->type = type;
- if (snapshot != NULL) {
- ao2_ref(snapshot, +1);
- entry->snapshot = snapshot;
+ if (is_remote) {
+ if (AST_VECTOR_APPEND(&entry->remote, snapshot)) {
+ ao2_cleanup(entry);
+ return NULL;
+ }
+ } else {
+ entry->local = snapshot;
}
+ ao2_bump(snapshot);
- ao2_ref(entry, +1);
return entry;
}
static int cache_entry_hash(const void *obj, int flags)
{
- const struct cache_entry *entry = obj;
+ const struct stasis_cache_entry *object;
+ const struct cache_entry_key *key;
int hash = 0;
- ast_assert(!(flags & OBJ_KEY));
+ switch (flags & OBJ_SEARCH_MASK) {
+ case OBJ_SEARCH_KEY:
+ key = obj;
+ break;
+ case OBJ_SEARCH_OBJECT:
+ object = obj;
+ key = &object->key;
+ break;
+ default:
+ /* Hash can only work on something with a full key. */
+ ast_assert(0);
+ return 0;
+ }
- hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
- hash += ast_hashtab_hash_string(entry->id);
+ hash += ast_hashtab_hash_string(stasis_message_type_name(key->type));
+ hash += ast_hashtab_hash_string(key->id);
return hash;
}
static int cache_entry_cmp(void *obj, void *arg, int flags)
{
- const struct cache_entry *left = obj;
- const struct cache_entry *right = arg;
+ const struct stasis_cache_entry *object_left = obj;
+ const struct stasis_cache_entry *object_right = arg;
+ const struct cache_entry_key *right_key = obj;
+ int cmp;
+
+ switch (flags & OBJ_SEARCH_MASK) {
+ case OBJ_SEARCH_OBJECT:
+ right_key = &object_right->key;
+ /* Fall through */
+ case OBJ_SEARCH_KEY:
+ cmp = object_left->key.type != right_key->type
+ || strcmp(object_left->key.id, right_key->id);
+ break;
+ case OBJ_SEARCH_PARTIAL_KEY:
+ /* Not supported by container */
+ ast_assert(0);
+ cmp = -1;
+ break;
+ default:
+ /*
+ * What arg points to is specific to this traversal callback
+ * and has no special meaning to astobj2.
+ */
+ cmp = 0;
+ break;
+ }
+ if (cmp) {
+ return 0;
+ }
+ /*
+ * At this point the traversal callback is identical to a sorted
+ * container.
+ */
+ return CMP_MATCH;
+}
+
+static void cache_dtor(void *obj)
+{
+ struct stasis_cache *cache = obj;
- ast_assert(!(flags & OBJ_KEY));
+ ao2_cleanup(cache->entries);
+ cache->entries = NULL;
+}
- if (left->type == right->type && strcmp(left->id, right->id) == 0) {
- return CMP_MATCH | CMP_STOP;
+struct stasis_cache *stasis_cache_create_full(snapshot_get_id id_fn,
+ cache_aggregate_calc_fn aggregate_calc_fn,
+ cache_aggregate_publish_fn aggregate_publish_fn)
+{
+ struct stasis_cache *cache;
+
+ cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
+ AO2_ALLOC_OPT_LOCK_NOLOCK);
+ if (!cache) {
+ return NULL;
}
- return 0;
+ cache->entries = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
+ NUM_CACHE_BUCKETS, cache_entry_hash, NULL, cache_entry_cmp);
+ if (!cache->entries) {
+ ao2_cleanup(cache);
+ return NULL;
+ }
+
+ cache->id_fn = id_fn;
+ cache->aggregate_calc_fn = aggregate_calc_fn;
+ cache->aggregate_publish_fn = aggregate_publish_fn;
+
+ return cache;
}
-static void cache_dtor(void *obj)
+struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
{
- struct stasis_cache *cache = obj;
+ return stasis_cache_create_full(id_fn, NULL, NULL);
+}
- ao2_cleanup(cache->entries);
- cache->entries = NULL;
+struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry)
+{
+ return entry->aggregate;
}
-struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
+struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry)
{
- RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
+ return entry->local;
+}
- cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
- AO2_ALLOC_OPT_LOCK_NOLOCK);
- if (!cache) {
- return NULL;
- }
+struct stasis_message *stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx)
+{
+ if (idx < AST_VECTOR_SIZE(&entry->remote)) {
+ return AST_VECTOR_GET(&entry->remote, idx);
+ }
+ return NULL;
+}
- cache->entries = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash,
- cache_entry_cmp);
- if (!cache->entries) {
- return NULL;
- }
+/*!
+ * \internal
+ * \brief Find the cache entry in the cache entries container.
+ *
+ * \param entries Container of cached entries.
+ * \param type Type of message to retrieve the cache entry.
+ * \param id Identity of the snapshot to retrieve the cache entry.
+ *
+ * \note The entries container is already locked.
+ *
+ * \retval Cache-entry on success.
+ * \retval NULL Not in cache.
+ */
+static struct stasis_cache_entry *cache_find(struct ao2_container *entries, struct stasis_message_type *type, const char *id)
+{
+ struct cache_entry_key search_key;
+
+ search_key.type = type;
+ search_key.id = id;
+ return ao2_find(entries, &search_key, OBJ_SEARCH_KEY | OBJ_NOLOCK);
+}
+
+/*!
+ * \internal
+ * \brief Remove the stasis snapshot in the cache entry determined by eid.
+ *
+ * \param entries Container of cached entries.
+ * \param cached_entry The entry to remove the snapshot from.
+ * \param eid Which snapshot in the cached entry.
+ *
+ * \note The entries container is already locked.
+ *
+ * \return Previous stasis entry snapshot.
+ */
+static struct stasis_message *cache_remove(struct ao2_container *entries, struct stasis_cache_entry *cached_entry, const struct ast_eid *eid)
+{
+ struct stasis_message *old_snapshot;
+ int is_remote;
+
+ is_remote = ast_eid_cmp(eid, &ast_eid_default);
+ if (!is_remote) {
+ old_snapshot = cached_entry->local;
+ cached_entry->local = NULL;
+ } else {
+ int idx;
+
+ old_snapshot = NULL;
+ for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
+ struct stasis_message *cur;
+
+ cur = AST_VECTOR_GET(&cached_entry->remote, idx);
+ if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
+ old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
+ break;
+ }
+ }
+ }
+
+ if (!cached_entry->local && !AST_VECTOR_SIZE(&cached_entry->remote)) {
+ ao2_unlink_flags(entries, cached_entry, OBJ_NOLOCK);
+ }
+
+ return old_snapshot;
+}
- cache->id_fn = id_fn;
+/*!
+ * \internal
+ * \brief Update the stasis snapshot in the cache entry determined by eid.
+ *
+ * \param cached_entry The entry to remove the snapshot from.
+ * \param eid Which snapshot in the cached entry.
+ * \param new_snapshot Snapshot to replace the old snapshot.
+ *
+ * \return Previous stasis entry snapshot.
+ */
+static struct stasis_message *cache_udpate(struct stasis_cache_entry *cached_entry, const struct ast_eid *eid, struct stasis_message *new_snapshot)
+{
+ struct stasis_message *old_snapshot;
+ int is_remote;
+ int idx;
+
+ is_remote = ast_eid_cmp(eid, &ast_eid_default);
+ if (!is_remote) {
+ old_snapshot = cached_entry->local;
+ cached_entry->local = ao2_bump(new_snapshot);
+ return old_snapshot;
+ }
+
+ old_snapshot = NULL;
+ for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
+ struct stasis_message *cur;
- ao2_ref(cache, +1);
- return cache;
+ cur = AST_VECTOR_GET(&cached_entry->remote, idx);
+ if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
+ old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
+ break;
+ }
+ }
+ if (!AST_VECTOR_APPEND(&cached_entry->remote, new_snapshot)) {
+ ao2_bump(new_snapshot);
+ }
+
+ return old_snapshot;
}
-static struct stasis_message *cache_put(struct stasis_cache *cache,
- struct stasis_message_type *type, const char *id,
+struct cache_put_snapshots {
+ /*! Old cache eid snapshot. */
+ struct stasis_message *old;
+ /*! Old cache aggregate snapshot. */
+ struct stasis_message *aggregate_old;
+ /*! New cache aggregate snapshot. */
+ struct stasis_message *aggregate_new;
+};
+
+static struct cache_put_snapshots cache_put(struct stasis_cache *cache,
+ struct stasis_message_type *type, const char *id, const struct ast_eid *eid,
struct stasis_message *new_snapshot)
{
- RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
- RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
- struct stasis_message *old_snapshot = NULL;
+ struct stasis_cache_entry *cached_entry;
+ struct cache_put_snapshots snapshots;
ast_assert(cache->entries != NULL);
+ ast_assert(eid != NULL);/* Aggregate snapshots not allowed to be put directly. */
ast_assert(new_snapshot == NULL ||
type == stasis_message_type(new_snapshot));
- new_entry = cache_entry_create(type, id, new_snapshot);
+ memset(&snapshots, 0, sizeof(snapshots));
+
+ ao2_wrlock(cache->entries);
+
+ cached_entry = cache_find(cache->entries, type, id);
- if (new_snapshot == NULL) {
- /* Remove entry from cache */
- cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_UNLINK);
+ /* Update the eid snapshot. */
+ if (!new_snapshot) {
+ /* Remove snapshot from cache */
if (cached_entry) {
- old_snapshot = cached_entry->snapshot;
- cached_entry->snapshot = NULL;
+ snapshots.old = cache_remove(cache->entries, cached_entry, eid);
}
+ } else if (cached_entry) {
+ /* Update snapshot in cache */
+ snapshots.old = cache_udpate(cached_entry, eid, new_snapshot);
} else {
- /* Insert/update cache */
- SCOPED_AO2LOCK(lock, cache->entries);
-
- cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_NOLOCK);
+ /* Insert into the cache */
+ cached_entry = cache_entry_create(type, id, new_snapshot);
if (cached_entry) {
- /* Update cache. Because objects are moving, no need to update refcounts. */
- old_snapshot = cached_entry->snapshot;
- cached_entry->snapshot = new_entry->snapshot;
- new_entry->snapshot = NULL;
- } else {
- /* Insert into the cache */
- ao2_link_flags(cache->entries, new_entry, OBJ_NOLOCK);
+ ao2_link_flags(cache->entries, cached_entry, OBJ_NOLOCK);
}
+ }
+ /* Update the aggregate snapshot. */
+ if (cache->aggregate_calc_fn && cached_entry) {
+ snapshots.aggregate_new = cache->aggregate_calc_fn(cached_entry, new_snapshot);
+ snapshots.aggregate_old = cached_entry->aggregate;
+ cached_entry->aggregate = ao2_bump(snapshots.aggregate_new);
}
- return old_snapshot;
+ ao2_unlock(cache->entries);
+
+ ao2_cleanup(cached_entry);
+ return snapshots;
}
-struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
+/*!
+ * \internal
+ * \brief Dump all entity snapshots in the cache entry into the given container.
+ *
+ * \param snapshots Container to put all snapshots in the cache entry.
+ * \param entry Cache entry to use.
+ *
+ * \retval 0 on success.
+ * \retval non-zero on error.
+ */
+static int cache_entry_dump(struct ao2_container *snapshots, const struct stasis_cache_entry *entry)
+{
+ int idx;
+ int err = 0;
+
+ ast_assert(snapshots != NULL);
+ ast_assert(entry != NULL);
+
+ /* The aggregate snapshot is not a snapshot from an entity. */
+
+ if (entry->local) {
+ err |= !ao2_link(snapshots, entry->local);
+ }
+
+ for (idx = 0; !err && idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
+ struct stasis_message *snapshot;
+
+ snapshot = AST_VECTOR_GET(&entry->remote, idx);
+ err |= !ao2_link(snapshots, snapshot);
+ }
+
+ return err;
+}
+
+struct ao2_container *stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
{
- RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
- RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
+ struct stasis_cache_entry *cached_entry;
+ struct ao2_container *found;
+ ast_assert(cache != NULL);
ast_assert(cache->entries != NULL);
+ ast_assert(type != NULL);
+ ast_assert(id != NULL);
- search_entry = cache_entry_create(type, id, NULL);
- if (search_entry == NULL) {
+ found = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
+ if (!found) {
return NULL;
}
- cached_entry = ao2_find(cache->entries, search_entry, OBJ_POINTER);
- if (cached_entry == NULL) {
- return NULL;
+ ao2_rdlock(cache->entries);
+
+ cached_entry = cache_find(cache->entries, type, id);
+ if (cached_entry && cache_entry_dump(found, cached_entry)) {
+ ao2_cleanup(found);
+ found = NULL;
+ }
+
+ ao2_unlock(cache->entries);
+
+ ao2_cleanup(cached_entry);
+ return found;
+}
+
+/*!
+ * \internal
+ * \brief Retrieve an item from the cache entry for a specific eid.
+ *
+ * \param entry Cache entry to use.
+ * \param eid Specific entity id to retrieve. NULL for aggregate.
+ *
+ * \note The returned snapshot has not had its reference bumped.
+ *
+ * \retval Snapshot from the cache.
+ * \retval \c NULL if snapshot is not found.
+ */
+static struct stasis_message *cache_entry_by_eid(const struct stasis_cache_entry *entry, const struct ast_eid *eid)
+{
+ int is_remote;
+ int idx;
+
+ if (!eid) {
+ /* Get aggregate. */
+ return entry->aggregate;
+ }
+
+ /* Get snapshot with specific eid. */
+ is_remote = ast_eid_cmp(eid, &ast_eid_default);
+ if (!is_remote) {
+ return entry->local;
}
- ast_assert(cached_entry->snapshot != NULL);
- ao2_ref(cached_entry->snapshot, +1);
- return cached_entry->snapshot;
+ for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
+ struct stasis_message *cur;
+
+ cur = AST_VECTOR_GET(&entry->remote, idx);
+ if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
+ return cur;
+ }
+ }
+
+ return NULL;
+}
+
+struct stasis_message *stasis_cache_get_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const char *id, const struct ast_eid *eid)
+{
+ struct stasis_cache_entry *cached_entry;
+ struct stasis_message *snapshot = NULL;
+
+ ast_assert(cache != NULL);
+ ast_assert(cache->entries != NULL);
+ ast_assert(type != NULL);
+ ast_assert(id != NULL);
+
+ ao2_rdlock(cache->entries);
+
+ cached_entry = cache_find(cache->entries, type, id);
+ if (cached_entry) {
+ snapshot = cache_entry_by_eid(cached_entry, eid);
+ ao2_bump(snapshot);
+ }
+
+ ao2_unlock(cache->entries);
+
+ ao2_cleanup(cached_entry);
+ return snapshot;
+}
+
+struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
+{
+ return stasis_cache_get_by_eid(cache, type, id, &ast_eid_default);
}
struct cache_dump_data {
- struct ao2_container *cached;
+ struct ao2_container *container;
struct stasis_message_type *type;
+ const struct ast_eid *eid;
};
-static int cache_dump_cb(void *obj, void *arg, int flags)
+static int cache_dump_by_eid_cb(void *obj, void *arg, int flags)
{
struct cache_dump_data *cache_dump = arg;
- struct cache_entry *entry = obj;
+ struct stasis_cache_entry *entry = obj;
+
+ if (!cache_dump->type || entry->key.type == cache_dump->type) {
+ struct stasis_message *snapshot;
- if (!cache_dump->type || entry->type == cache_dump->type) {
- ao2_link(cache_dump->cached, entry->snapshot);
+ snapshot = cache_entry_by_eid(entry, cache_dump->eid);
+ if (snapshot) {
+ if (!ao2_link(cache_dump->container, snapshot)) {
+ ao2_cleanup(cache_dump->container);
+ cache_dump->container = NULL;
+ return CMP_STOP;
+ }
+ }
}
return 0;
}
+struct ao2_container *stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
+{
+ struct cache_dump_data cache_dump;
+
+ ast_assert(cache != NULL);
+ ast_assert(cache->entries != NULL);
+
+ cache_dump.eid = eid;
+ cache_dump.type = type;
+ cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
+ if (!cache_dump.container) {
+ return NULL;
+ }
+
+ ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_by_eid_cb, &cache_dump);
+ return cache_dump.container;
+}
+
struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
{
+ return stasis_cache_dump_by_eid(cache, type, &ast_eid_default);
+}
+
+static int cache_dump_all_cb(void *obj, void *arg, int flags)
+{
+ struct cache_dump_data *cache_dump = arg;
+ struct stasis_cache_entry *entry = obj;
+
+ if (!cache_dump->type || entry->key.type == cache_dump->type) {
+ if (cache_entry_dump(cache_dump->container, entry)) {
+ ao2_cleanup(cache_dump->container);
+ cache_dump->container = NULL;
+ return CMP_STOP;
+ }
+ }
+
+ return 0;
+}
+
+struct ao2_container *stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type)
+{
struct cache_dump_data cache_dump;
+ ast_assert(cache != NULL);
ast_assert(cache->entries != NULL);
+ cache_dump.eid = NULL;
cache_dump.type = type;
- cache_dump.cached = ao2_container_alloc_options(
- AO2_ALLOC_OPT_LOCK_NOLOCK, 1, NULL, NULL);
- if (!cache_dump.cached) {
+ cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
+ if (!cache_dump.container) {
return NULL;
}
- ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump);
- return cache_dump.cached;
+ ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_all_cb, &cache_dump);
+ return cache_dump.container;
}
STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
@@ -380,9 +762,13 @@ static struct stasis_message *update_create(struct stasis_message *old_snapshot,
static void caching_topic_exec(void *data, struct stasis_subscription *sub,
struct stasis_message *message)
{
- RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
+ struct stasis_caching_topic *caching_topic_needs_unref;
struct stasis_caching_topic *caching_topic = data;
- const char *id = NULL;
+ struct stasis_message *msg;
+ struct stasis_message *msg_put;
+ struct stasis_message_type *msg_type;
+ const struct ast_eid *msg_eid;
+ const char *msg_id;
ast_assert(caching_topic != NULL);
ast_assert(caching_topic->topic != NULL);
@@ -391,50 +777,62 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub,
if (stasis_subscription_final_message(sub, message)) {
caching_topic_needs_unref = caching_topic;
+ } else {
+ caching_topic_needs_unref = NULL;
}
- /* Handle cache clear event */
- if (stasis_cache_clear_type() == stasis_message_type(message)) {
- RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
- struct stasis_message *clear_msg = stasis_message_data(message);
- const char *clear_id = caching_topic->cache->id_fn(clear_msg);
- struct stasis_message_type *clear_type = stasis_message_type(clear_msg);
+ msg_type = stasis_message_type(message);
+ if (stasis_cache_clear_type() == msg_type) {
+ /* Cache clear event. */
+ msg_put = NULL;
+ msg = stasis_message_data(message);
+ msg_type = stasis_message_type(msg);
+ } else {
+ /* Normal cache update event. */
+ msg_put = message;
+ msg = message;
+ }
+ ast_assert(msg_type != NULL);
- ast_assert(clear_type != NULL);
+ msg_eid = stasis_message_eid(msg);/* msg_eid is NULL for aggregate message. */
+ msg_id = caching_topic->cache->id_fn(msg);
+ if (msg_id && msg_eid) {
+ struct stasis_message *update;
+ struct cache_put_snapshots snapshots;
- if (clear_id) {
- old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL);
- if (old_snapshot) {
- update = update_create(old_snapshot, NULL);
+ /* Update the cache */
+ snapshots = cache_put(caching_topic->cache, msg_type, msg_id, msg_eid, msg_put);
+ if (snapshots.old || msg_put) {
+ update = update_create(snapshots.old, msg_put);
+ if (update) {
stasis_publish(caching_topic->topic, update);
- return;
}
-
+ ao2_cleanup(update);
+ } else {
ast_log(LOG_ERROR,
"Attempting to remove an item from the %s cache that isn't there: %s %s\n",
- stasis_topic_name(caching_topic->topic), stasis_message_type_name(clear_type), clear_id);
- return;
+ stasis_topic_name(caching_topic->topic),
+ stasis_message_type_name(msg_type), msg_id);
}
- }
-
- id = caching_topic->cache->id_fn(message);
- if (id == NULL) {
- /* Object isn't cached; discard */
- } else {
- /* Update the cache */
- RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
- RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
- old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message);
-
- update = update_create(old_snapshot, message);
- if (update == NULL) {
- return;
+ if (snapshots.aggregate_old != snapshots.aggregate_new) {
+ if (snapshots.aggregate_new && caching_topic->cache->aggregate_publish_fn) {
+ caching_topic->cache->aggregate_publish_fn(caching_topic->original_topic,
+ snapshots.aggregate_new);
+ }
+ update = update_create(snapshots.aggregate_old, snapshots.aggregate_new);
+ if (update) {
+ stasis_publish(caching_topic->topic, update);
+ }
+ ao2_cleanup(update);
}
- stasis_publish(caching_topic->topic, update);
+ ao2_cleanup(snapshots.old);
+ ao2_cleanup(snapshots.aggregate_old);
+ ao2_cleanup(snapshots.aggregate_new);
}
+
+ ao2_cleanup(caching_topic_needs_unref);
}
struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-11 17:52:37 +0000