Stasis Message API. More...

#include "asterisk.h"
#include "asterisk/astobj2.h"
#include "asterisk/hashtab.h"
#include "asterisk/stasis_internal.h"
#include "asterisk/stasis.h"
#include "asterisk/utils.h"
#include "asterisk/vector.h"

Include dependency graph for stasis_cache.c:

Go to the source code of this file.

Data Structures
struct	cache_dump_data

struct	cache_entry_key
	The key for an entry in the cache. More...

struct	cache_put_snapshots

struct	stasis_cache

struct	stasis_cache_entry

struct	stasis_caching_topic

Macros
#define	NUM_CACHE_BUCKETS 563

Functions
static void	cache_dtor (void *obj)

static int	cache_dump_all_cb (void obj, void arg, int flags)

static int	cache_dump_by_eid_cb (void obj, void arg, int flags)

static struct stasis_message *	cache_entry_by_eid (const struct stasis_cache_entry entry, const struct ast_eid eid)

static int	cache_entry_cmp (void obj, void arg, int flags)

static void	cache_entry_compute_hash (struct cache_entry_key *key)

static struct stasis_cache_entry *	cache_entry_create (struct stasis_message_type type, const char id, struct stasis_message *snapshot)

static void	cache_entry_dtor (void *obj)

static int	cache_entry_dump (struct ao2_container snapshots, const struct stasis_cache_entry entry)

static int	cache_entry_hash (const void *obj, int flags)

static struct stasis_cache_entry *	cache_find (struct ao2_container entries, struct stasis_message_type type, const char *id)

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)

static struct stasis_message *	cache_remove (struct ao2_container entries, struct stasis_cache_entry cached_entry, const struct ast_eid *eid)

static struct stasis_message *	cache_udpate (struct stasis_cache_entry cached_entry, const struct ast_eid eid, struct stasis_message *new_snapshot)

static void	caching_topic_exec (void data, struct stasis_subscription sub, struct stasis_message *message)

static void	print_cache_entry (void v_obj, void where, ao2_prnt_fn *prnt)

static void	stasis_cache_cleanup (void)

struct stasis_message *	stasis_cache_clear_create (struct stasis_message *id_message)
	A message which instructs the caching topic to remove an entry from its cache. More...

struct stasis_cache *	stasis_cache_create (snapshot_get_id id_fn)
	Create a cache. More...

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)
	Create a cache. More...

struct ao2_container *	stasis_cache_dump (struct stasis_cache cache, struct stasis_message_type type)
	Dump cached items to a subscription for the ast_eid_default entity. More...

struct ao2_container *	stasis_cache_dump_all (struct stasis_cache cache, struct stasis_message_type type)
	Dump all entity items from the cache to a subscription. More...

struct ao2_container *	stasis_cache_dump_by_eid (struct stasis_cache cache, struct stasis_message_type type, const struct ast_eid *eid)
	Dump cached items to a subscription for a specific entity. More...

struct stasis_message *	stasis_cache_entry_get_aggregate (struct stasis_cache_entry *entry)
	Get the aggregate cache entry snapshot. More...

struct stasis_message *	stasis_cache_entry_get_local (struct stasis_cache_entry *entry)
	Get the local entity's cache entry snapshot. More...

struct stasis_message *	stasis_cache_entry_get_remote (struct stasis_cache_entry *entry, int idx)
	Get a remote entity's cache entry snapshot by index. More...

struct stasis_message *	stasis_cache_get (struct stasis_cache cache, struct stasis_message_type type, const char *id)
	Retrieve an item from the cache for the ast_eid_default entity. More...

struct ao2_container *	stasis_cache_get_all (struct stasis_cache cache, struct stasis_message_type type, const char *id)
	Retrieve all matching entity items from the cache. More...

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)
	Retrieve an item from the cache for a specific entity. More...

int	stasis_cache_init (void)

static void	stasis_cache_update_dtor (void *obj)

int	stasis_caching_accept_message_type (struct stasis_caching_topic caching_topic, struct stasis_message_type type)
	Indicate to a caching topic that we are interested in a message type. More...

struct stasis_topic *	stasis_caching_get_topic (struct stasis_caching_topic *caching_topic)
	Returns the topic of cached events from a caching topics. More...

int	stasis_caching_set_filter (struct stasis_caching_topic *caching_topic, enum stasis_subscription_message_filter filter)
	Set the message type filtering level on a cache. More...

struct stasis_caching_topic *	stasis_caching_topic_create (struct stasis_topic original_topic, struct stasis_cache cache)
	Create a topic which monitors and caches messages from another topic. More...

static void	stasis_caching_topic_dtor (void *obj)

struct stasis_caching_topic *	stasis_caching_unsubscribe (struct stasis_caching_topic *caching_topic)
	Unsubscribes a caching topic from its upstream topic. More...

struct stasis_caching_topic *	stasis_caching_unsubscribe_and_join (struct stasis_caching_topic *caching_topic)
	Unsubscribes a caching topic from its upstream topic, blocking until all messages have been forwarded. More...

	STASIS_MESSAGE_TYPE_DEFN (stasis_cache_clear_type)

	STASIS_MESSAGE_TYPE_DEFN (stasis_cache_update_type)

static struct stasis_message *	update_create (struct stasis_message old_snapshot, struct stasis_message new_snapshot)

Detailed Description

Stasis Message API.

Author: David M. Lee, II dlee@.nosp@m.digi.nosp@m.um.co.nosp@m.m

Definition in file stasis_cache.c.

Macro Definition Documentation

◆ NUM_CACHE_BUCKETS

#define NUM_CACHE_BUCKETS 563

Definition at line 42 of file stasis_cache.c.

Referenced by stasis_cache_create_full().

Function Documentation

◆ cache_dtor()

static void cache_dtor ( void * obj )

static

Definition at line 326 of file stasis_cache.c.

References ao2_cleanup, cache, stasis_cache::entries, and NULL.

Referenced by stasis_cache_create_full().

 {
    struct stasis_cache *cache = obj;
 
    ao2_cleanup(cache->entries);
    cache->entries = NULL;
 }

◆ cache_dump_all_cb()

static int cache_dump_all_cb	(	void *	obj,
		void *	arg,
		int	flags
	)

static

Definition at line 741 of file stasis_cache.c.

References ao2_cleanup, cache_entry_dump(), CMP_STOP, cache_dump_data::container, stasis_cache_entry::key, NULL, cache_entry_key::type, and cache_dump_data::type.

Referenced by stasis_cache_dump_all().

 {
    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;
 }

◆ cache_dump_by_eid_cb()

static int cache_dump_by_eid_cb	(	void *	obj,
		void *	arg,
		int	flags
	)

static

Definition at line 697 of file stasis_cache.c.

References ao2_cleanup, ao2_link, cache_entry_by_eid(), CMP_STOP, cache_dump_data::container, cache_dump_data::eid, stasis_cache_entry::key, NULL, cache_entry_key::type, and cache_dump_data::type.

Referenced by stasis_cache_dump_by_eid().

 {
    struct cache_dump_data *cache_dump = arg;
    struct stasis_cache_entry *entry = obj;
 
    if (!cache_dump->type || entry->key.type == cache_dump->type) {
       struct stasis_message *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;
 }

◆ cache_entry_by_eid()

static struct stasis_message* cache_entry_by_eid	(	const struct stasis_cache_entry *	entry,
		const struct ast_eid *	eid
	)

static

Definition at line 631 of file stasis_cache.c.

References stasis_cache_entry::aggregate, ast_eid_cmp(), ast_eid_default, AST_VECTOR_GET, AST_VECTOR_SIZE, stasis_cache_entry::local, NULL, and stasis_message_eid().

Referenced by cache_dump_by_eid_cb(), and stasis_cache_get_by_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;
    }
 
    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;
 }

◆ cache_entry_cmp()

static int cache_entry_cmp	(	void *	obj,
		void *	arg,
		int	flags
	)

static

Definition at line 288 of file stasis_cache.c.

References ast_assert, CMP_MATCH, cache_entry_key::id, stasis_cache_entry::key, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, OBJ_SEARCH_PARTIAL_KEY, and cache_entry_key::type.

Referenced by stasis_cache_create_full().

 {
    const struct stasis_cache_entry *object_left = obj;
    const struct stasis_cache_entry *object_right = arg;
    const struct cache_entry_key *right_key = arg;
    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;
 }

◆ cache_entry_compute_hash()

static void cache_entry_compute_hash ( struct cache_entry_key * key )

static

Definition at line 206 of file stasis_cache.c.

References ast_hashtab_hash_string(), cache_entry_key::hash, cache_entry_key::id, stasis_message_type_hash(), and cache_entry_key::type.

Referenced by cache_entry_create(), and cache_find().

 {
    key->hash = stasis_message_type_hash(key->type);
    key->hash += ast_hashtab_hash_string(key->id);
 }

◆ cache_entry_create()

static struct stasis_cache_entry* cache_entry_create	(	struct stasis_message_type *	type,
		const char *	id,
		struct stasis_message *	snapshot
	)

static

Definition at line 212 of file stasis_cache.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_bump, ao2_cleanup, ast_assert, ast_eid_cmp(), ast_eid_default, ast_strdup, AST_VECTOR_APPEND, AST_VECTOR_INIT, cache_entry_compute_hash(), cache_entry_dtor(), cache_entry_key::id, stasis_cache_entry::key, stasis_cache_entry::local, NULL, stasis_message_eid(), stasis_message_type(), type, and cache_entry_key::type.

Referenced by cache_put().

 {
    struct stasis_cache_entry *entry;
    int is_remote;
 
    ast_assert(id != NULL);
    ast_assert(snapshot != NULL);
 
    if (!type) {
       return NULL;
    }
 
    entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
       AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (!entry) {
       return NULL;
    }
 
    entry->key.id = ast_strdup(id);
    if (!entry->key.id) {
       ao2_cleanup(entry);
       return NULL;
    }
    /*
     * Normal ao2 ref counting rules says we should increment the message
     * type ref here and decrement it in cache_entry_dtor().  However, the
     * stasis message snapshot is cached here, will always have the same type
     * as the cache entry, and can legitimately cause the type ref count to
     * hit the excessive ref count assertion.  Since the cache entry will
     * always have a snapshot we can get away with not holding a ref here.
     */
    ast_assert(type == stasis_message_type(snapshot));
    entry->key.type = type;
    cache_entry_compute_hash(&entry->key);
 
    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;
    }
 
    if (is_remote) {
       if (AST_VECTOR_APPEND(&entry->remote, snapshot)) {
          ao2_cleanup(entry);
          return NULL;
       }
    } else {
       entry->local = snapshot;
    }
    ao2_bump(snapshot);
 
    return entry;
 }

◆ cache_entry_dtor()

static void cache_entry_dtor ( void * obj )

static

Definition at line 183 of file stasis_cache.c.

References stasis_cache_entry::aggregate, ao2_cleanup, ast_free, AST_VECTOR_FREE, AST_VECTOR_GET, AST_VECTOR_SIZE, cache_entry_key::id, stasis_cache_entry::key, stasis_cache_entry::local, NULL, and cache_entry_key::type.

Referenced by cache_entry_create().

 {
    struct stasis_cache_entry *entry = obj;
    size_t idx;
 
    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);
 }

◆ cache_entry_dump()

static int cache_entry_dump	(	struct ao2_container *	snapshots,
		const struct stasis_cache_entry *	entry
	)

static

Definition at line 563 of file stasis_cache.c.

References ao2_link, ast_assert, AST_VECTOR_GET, AST_VECTOR_SIZE, stasis_cache_entry::local, and NULL.

Referenced by cache_dump_all_cb(), and stasis_cache_get_all().

 {
    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;
 }

◆ cache_entry_hash()

static int cache_entry_hash	(	const void *	obj,
		int	flags
	)

static

Definition at line 266 of file stasis_cache.c.

References ast_assert, cache_entry_key::hash, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, and OBJ_SEARCH_OBJECT.

Referenced by stasis_cache_create_full().

 {
    const struct stasis_cache_entry *object;
    const struct cache_entry_key *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;
    }
 
    return (int)key->hash;
 }

◆ cache_find()

static struct stasis_cache_entry* cache_find	(	struct ao2_container *	entries,
		struct stasis_message_type *	type,
		const char *	id
	)

static

Definition at line 396 of file stasis_cache.c.

References ao2_find, ast_assert, cache_entry_compute_hash(), cache_entry_key::id, id, stasis_cache_entry::key, OBJ_NOLOCK, OBJ_SEARCH_KEY, stasis_message_type_name(), type, and cache_entry_key::type.

Referenced by cache_put(), caching_topic_exec(), stasis_cache_get_all(), and stasis_cache_get_by_eid().

 {
    struct cache_entry_key search_key;
    struct stasis_cache_entry *entry;
 
    search_key.type = type;
    search_key.id = id;
    cache_entry_compute_hash(&search_key);
    entry = ao2_find(entries, &search_key, OBJ_SEARCH_KEY | OBJ_NOLOCK);
 
    /* Ensure that what we looked for is what we found. */
    ast_assert(!entry
       || (!strcmp(stasis_message_type_name(entry->key.type),
          stasis_message_type_name(type)) && !strcmp(entry->key.id, id)));
    return entry;
 }

◆ cache_put()

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
	)

static

Definition at line 505 of file stasis_cache.c.

References stasis_cache_entry::aggregate, cache_put_snapshots::aggregate_new, cache_put_snapshots::aggregate_old, ao2_bump, ao2_cleanup, ao2_link_flags, ao2_unlock, ao2_wrlock, ast_assert, cache_entry_create(), cache_find(), cache_remove(), cache_udpate(), NULL, OBJ_NOLOCK, cache_put_snapshots::old, and stasis_message_type().

Referenced by caching_topic_exec().

 {
    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));
 
    memset(&snapshots, 0, sizeof(snapshots));
 
    ao2_wrlock(cache->entries);
 
    cached_entry = cache_find(cache->entries, type, id);
 
    /* Update the eid snapshot. */
    if (!new_snapshot) {
       /* Remove snapshot from cache */
       if (cached_entry) {
          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 into the cache */
       cached_entry = cache_entry_create(type, id, new_snapshot);
       if (cached_entry) {
          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);
    }
 
    ao2_unlock(cache->entries);
 
    ao2_cleanup(cached_entry);
    return snapshots;
 }

◆ cache_remove()

static struct stasis_message* cache_remove	(	struct ao2_container *	entries,
		struct stasis_cache_entry *	cached_entry,
		const struct ast_eid *	eid
	)

static

Definition at line 425 of file stasis_cache.c.

References ao2_unlink_flags, ast_eid_cmp(), ast_eid_default, AST_VECTOR_GET, AST_VECTOR_REMOVE_UNORDERED, AST_VECTOR_SIZE, stasis_cache_entry::local, NULL, OBJ_NOLOCK, and stasis_message_eid().

Referenced by cache_put(), and caching_topic_exec().

 {
    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_udpate()

static struct stasis_message* cache_udpate	(	struct stasis_cache_entry *	cached_entry,
		const struct ast_eid *	eid,
		struct stasis_message *	new_snapshot
	)

static

Definition at line 466 of file stasis_cache.c.

References ao2_bump, ast_eid_cmp(), ast_eid_default, AST_VECTOR_APPEND, AST_VECTOR_GET, AST_VECTOR_REMOVE_UNORDERED, AST_VECTOR_SIZE, stasis_cache_entry::local, NULL, and stasis_message_eid().

Referenced by cache_put().

 {
    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;
 
       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;
 }

◆ caching_topic_exec()

static void caching_topic_exec	(	void *	data,
		struct stasis_subscription *	sub,
		struct stasis_message *	message
	)

static

Definition at line 833 of file stasis_cache.c.

References stasis_cache::aggregate_publish_fn, ao2_cleanup, ao2_ref, ao2_unlock, ao2_wrlock, ast_assert, ast_debug, stasis_caching_topic::cache, cache_find(), cache_put(), cache_remove(), stasis_subscription_change::description, stasis_cache::entries, stasis_cache::id_fn, NULL, stasis_caching_topic::original_topic, stasis_cache_clear_type(), stasis_message_data(), stasis_message_eid(), stasis_message_type(), stasis_message_type_name(), stasis_publish(), stasis_subscription_change_type(), stasis_subscription_final_message(), stasis_topic_name(), stasis_topic_subscribers(), stasis_caching_topic::topic, stasis_subscription_change::uniqueid, update(), and update_create().

Referenced by stasis_caching_topic_create().

 {
    struct stasis_caching_topic *caching_topic_needs_unref;
    struct stasis_caching_topic *caching_topic = data;
    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);
    ast_assert(caching_topic->cache != NULL);
    ast_assert(caching_topic->cache->id_fn != NULL);
 
    if (stasis_subscription_final_message(sub, message)) {
       caching_topic_needs_unref = caching_topic;
    } else {
       caching_topic_needs_unref = NULL;
    }
 
    msg_type = stasis_message_type(message);
 
    if (stasis_subscription_change_type() == msg_type) {
       struct stasis_subscription_change *change = stasis_message_data(message);
 
       /*
        * If this change type is an unsubscribe, we need to find the original
        * subscribe and remove it from the cache otherwise the cache will
        * continue to grow unabated.
        */
       if (strcmp(change->description, "Unsubscribe") == 0) {
          struct stasis_cache_entry *cached_sub;
 
          ao2_wrlock(caching_topic->cache->entries);
          cached_sub = cache_find(caching_topic->cache->entries, stasis_subscription_change_type(), change->uniqueid);
          if (cached_sub) {
             ao2_cleanup(cache_remove(caching_topic->cache->entries, cached_sub, stasis_message_eid(message)));
             ao2_cleanup(cached_sub);
          }
          ao2_unlock(caching_topic->cache->entries);
          ao2_cleanup(caching_topic_needs_unref);
          return;
       }
       msg_put = message;
       msg = message;
    } else 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);
 
    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;
 
       /* Update the cache */
       snapshots = cache_put(caching_topic->cache, msg_type, msg_id, msg_eid, msg_put);
       if (snapshots.old || msg_put) {
          if (stasis_topic_subscribers(caching_topic->topic)) {
             update = update_create(snapshots.old, msg_put);
             if (update) {
                stasis_publish(caching_topic->topic, update);
                ao2_ref(update, -1);
             }
          }
       } else {
          ast_debug(1,
             "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(msg_type), msg_id);
       }
 
       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);
          }
          if (stasis_topic_subscribers(caching_topic->topic)) {
             update = update_create(snapshots.aggregate_old, snapshots.aggregate_new);
             if (update) {
                stasis_publish(caching_topic->topic, update);
                ao2_ref(update, -1);
             }
          }
       }
 
       ao2_cleanup(snapshots.old);
       ao2_cleanup(snapshots.aggregate_old);
       ao2_cleanup(snapshots.aggregate_new);
    }
 
    ao2_cleanup(caching_topic_needs_unref);
 }

◆ print_cache_entry()

static void print_cache_entry	(	void *	v_obj,
		void *	where,
		ao2_prnt_fn *	prnt
	)

static

Definition at line 937 of file stasis_cache.c.

References cache_entry_key::hash, cache_entry_key::id, stasis_cache_entry::key, stasis_message_type_name(), and cache_entry_key::type.

Referenced by stasis_caching_topic_create().

 {
    struct stasis_cache_entry *entry = v_obj;
 
    if (!entry) {
       return;
    }
    prnt(where, "Type: %s  ID: %s  Hash: %u", stasis_message_type_name(entry->key.type),
       entry->key.id, entry->key.hash);
 }

◆ stasis_cache_cleanup()

static void stasis_cache_cleanup ( void )

static

Definition at line 1002 of file stasis_cache.c.

References stasis_cache_clear_type(), stasis_cache_update_type(), and STASIS_MESSAGE_TYPE_CLEANUP.

Referenced by stasis_cache_init().

 {
    STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
    STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
 }

◆ stasis_cache_clear_create()

struct stasis_message* stasis_cache_clear_create ( struct stasis_message * message )

A message which instructs the caching topic to remove an entry from its cache.

Parameters

message Message representative of the cache entry that should be cleared. This will become the data held in the stasis_cache_clear message.

Returns: Message which, when sent to a stasis_caching_topic, will clear the item from the cache.; NULL on error.

Since: 12

Definition at line 778 of file stasis_cache.c.

References stasis_cache_clear_type(), and stasis_message_create().

Referenced by ast_delete_mwi_state_full(), ast_device_state_clear_cache(), ast_endpoint_shutdown(), AST_TEST_DEFINE(), clear_node_cache(), and remove_device_states_cb().

 {
    return stasis_message_create(stasis_cache_clear_type(), id_message);
 }

◆ stasis_cache_create()

struct stasis_cache* stasis_cache_create ( snapshot_get_id id_fn )

Create a cache.

This is the backend store for a stasis_caching_topic. The cache is thread safe, allowing concurrent reads and writes.

The returned object is AO2 managed, so ao2_cleanup() when you're done.

Parameters

id_fn Callback to extract the id from a snapshot message.

Return values

New	cache indexed by id_fn.

Definition at line 360 of file stasis_cache.c.

References NULL, and stasis_cache_create_full().

Referenced by ast_presence_state_engine_init(), AST_TEST_DEFINE(), mwi_init(), and stasis_cp_all_create().

 {
    return stasis_cache_create_full(id_fn, NULL, NULL);
 }

◆ stasis_cache_create_full()

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
	)

Create a cache.

This is the backend store for a stasis_caching_topic. The cache is thread safe, allowing concurrent reads and writes.

The returned object is AO2 managed, so ao2_cleanup() when you're done.

Parameters

id_fn	Callback to extract the id from a snapshot message.
aggregate_calc_fn	Callback to calculate the aggregate cache entry.
aggregate_publish_fn	Callback to publish the aggregate cache entry.

Note: An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Return values

New	cache indexed by id_fn.

Definition at line 334 of file stasis_cache.c.

References stasis_cache::aggregate_calc_fn, stasis_cache::aggregate_publish_fn, AO2_ALLOC_OPT_LOCK_NOLOCK, AO2_ALLOC_OPT_LOCK_RWLOCK, ao2_alloc_options, ao2_cleanup, ao2_container_alloc_hash, cache, cache_dtor(), cache_entry_cmp(), cache_entry_hash(), stasis_cache::entries, stasis_cache::id_fn, NULL, and NUM_CACHE_BUCKETS.

Referenced by AST_TEST_DEFINE(), devstate_init(), and stasis_cache_create().

 {
    struct stasis_cache *cache;
 
    cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
       AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (!cache) {
       return NULL;
    }
 
    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;
 }

◆ stasis_cache_dump()

struct ao2_container* stasis_cache_dump	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type
	)

Dump cached items to a subscription for the ast_eid_default entity.

Parameters

cache	The cache to query.
type	Type of message to dump (any type if `NULL`).

Return values

ao2_container	containing all matches (must be unreffed by caller)

Definition at line 736 of file stasis_cache.c.

References ast_eid_default, and stasis_cache_dump_by_eid().

Referenced by action_presencestatelist(), ast_ari_endpoints_list(), ast_ari_endpoints_list_by_tech(), AST_TEST_DEFINE(), asterisk_publication_devicestate_refresh(), asterisk_publication_mwi_refresh(), asterisk_start_devicestate_publishing(), asterisk_start_mwi_publishing(), endpoints_scrape_cb(), load_module(), unload_module(), and xmpp_init_event_distribution().

 {
    return stasis_cache_dump_by_eid(cache, type, &ast_eid_default);
 }

◆ stasis_cache_dump_all()

struct ao2_container* stasis_cache_dump_all	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type
	)

Dump all entity items from the cache to a subscription.

Since: 12.2.0

Parameters

cache	The cache to query.
type	Type of message to dump (any type if `NULL`).

Return values

ao2_container	containing all matches (must be unreffed by caller)

Definition at line 757 of file stasis_cache.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, ast_assert, cache_dump_all_cb(), cache_dump_data::container, cache_dump_data::eid, stasis_cache::entries, NULL, OBJ_MULTIPLE, OBJ_NODATA, stasis_cache_clear_type(), stasis_cache_update_type(), STASIS_MESSAGE_TYPE_DEFN(), type, and cache_dump_data::type.

Referenced by AST_TEST_DEFINE(), cache_cleanup(), and cleanup_module().

 {
    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.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_all_cb, &cache_dump);
    return cache_dump.container;
 }

◆ stasis_cache_dump_by_eid()

struct ao2_container* stasis_cache_dump_by_eid	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type,
		const struct ast_eid *	eid
	)

Dump cached items to a subscription for a specific entity.

Since: 12.2.0

Parameters

cache	The cache to query.
type	Type of message to dump (any type if `NULL`).
eid	Specific entity id to retrieve. NULL for aggregate.

Return values

ao2_container	containing all matches (must be unreffed by caller)

Definition at line 718 of file stasis_cache.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, ast_assert, cache_dump_by_eid_cb(), cache_dump_data::container, cache_dump_data::eid, stasis_cache::entries, NULL, OBJ_MULTIPLE, OBJ_NODATA, type, and cache_dump_data::type.

Referenced by action_devicestatelist(), AST_TEST_DEFINE(), cpg_confchg_cb(), and stasis_cache_dump().

 {
    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;
 }

◆ stasis_cache_entry_get_aggregate()

struct stasis_message* stasis_cache_entry_get_aggregate ( struct stasis_cache_entry * entry )

Get the aggregate cache entry snapshot.

Since: 12.2.0

Parameters

entry Cache entry to get the aggregate snapshot.

Note: A reference is not given to the returned pointer so don't unref it.; An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Return values

Aggregate-snapshot	in cache.
NULL	if not present.

Definition at line 365 of file stasis_cache.c.

References stasis_cache_entry::aggregate.

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

 {
    return entry->aggregate;
 }

◆ stasis_cache_entry_get_local()

struct stasis_message* stasis_cache_entry_get_local ( struct stasis_cache_entry * entry )

Get the local entity's cache entry snapshot.

Since: 12.2.0

Parameters

entry Cache entry to get the local entity's snapshot.

Note: A reference is not given to the returned pointer so don't unref it.

Return values

Internal-snapshot	in cache.
NULL	if not present.

Definition at line 370 of file stasis_cache.c.

References stasis_cache_entry::local.

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

 {
    return entry->local;
 }

◆ stasis_cache_entry_get_remote()

struct stasis_message* stasis_cache_entry_get_remote	(	struct stasis_cache_entry *	entry,
		int	idx
	)

Get a remote entity's cache entry snapshot by index.

Since: 12.2.0

Parameters

entry	Cache entry to get a remote entity's snapshot.
idx	Which remote entity's snapshot to get.

Note: A reference is not given to the returned pointer so don't unref it.

Return values

Remote-entity-snapshot	in cache.
NULL	if not present.

Definition at line 375 of file stasis_cache.c.

References AST_VECTOR_GET, AST_VECTOR_SIZE, and NULL.

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

 {
    if (idx < AST_VECTOR_SIZE(&entry->remote)) {
       return AST_VECTOR_GET(&entry->remote, idx);
    }
    return NULL;
 }

◆ stasis_cache_get()

struct stasis_message* stasis_cache_get	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type,
		const char *	id
	)

Retrieve an item from the cache for the ast_eid_default entity.

The returned item is AO2 managed, so ao2_cleanup() when you're done with it.

Parameters

cache	The cache to query.
type	Type of message to retrieve.
id	Identity of the snapshot to retrieve.

Return values

Message	from the cache.

Definition at line 686 of file stasis_cache.c.

References ast_eid_default, and stasis_cache_get_by_eid().

Referenced by ast_endpoint_latest_snapshot(), AST_TEST_DEFINE(), get_cached_mwi(), has_voicemail(), presence_state_cached(), unistim_send_mwi_to_peer(), and update_registry().

 {
    return stasis_cache_get_by_eid(cache, type, id, &ast_eid_default);
 }

◆ stasis_cache_get_all()

struct ao2_container* stasis_cache_get_all	(	struct stasis_cache *	cache,
		struct stasis_message_type *	type,
		const char *	id
	)

Retrieve all matching entity items from the cache.

Since: 12.2.0

Parameters

cache	The cache to query.
type	Type of message to retrieve.
id	Identity of the snapshot to retrieve.

Return values

Container	of matching items found.

Definition at line 587 of file stasis_cache.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_cleanup, ao2_container_alloc_list, ao2_rdlock, ao2_unlock, ast_assert, cache_entry_dump(), cache_find(), stasis_cache::entries, and NULL.

Referenced by AST_TEST_DEFINE().

 {
    struct stasis_cache_entry *cached_entry;
    struct ao2_container *found;
 
    ast_assert(cache != NULL);
    ast_assert(cache->entries != NULL);
    ast_assert(id != NULL);
 
    if (!type) {
       return NULL;
    }
 
    found = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
    if (!found) {
       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;
 }

◆ stasis_cache_get_by_eid()

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
	)

Retrieve an item from the cache for a specific entity.

The returned item is AO2 managed, so ao2_cleanup() when you're done with it.

Parameters

cache	The cache to query.
type	Type of message to retrieve.
id	Identity of the snapshot to retrieve.
eid	Specific entity id to retrieve. NULL for aggregate.

Note: An aggregate message is a combined representation of the local and remote entities publishing the message data. e.g., An aggregate device state represents the combined device state from the local and any remote entities publishing state for a device. e.g., An aggregate MWI message is the old/new MWI counts accumulated from the local and any remote entities publishing to a mailbox.

Return values

Message	from the cache.

Definition at line 659 of file stasis_cache.c.

References ao2_bump, ao2_cleanup, ao2_rdlock, ao2_unlock, ast_assert, cache_entry_by_eid(), cache_find(), stasis_cache::entries, and NULL.

Referenced by ast_delete_mwi_state_full(), ast_device_state_clear_cache(), AST_TEST_DEFINE(), check_cache_aggregate(), devstate_cached(), and stasis_cache_get().

 {
    struct stasis_cache_entry *cached_entry;
    struct stasis_message *snapshot = NULL;
 
    ast_assert(cache != NULL);
    ast_assert(cache->entries != NULL);
    ast_assert(id != NULL);
 
    if (!type) {
       return 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;
 }

◆ stasis_cache_init()

int stasis_cache_init ( void )

Definition at line 1008 of file stasis_cache.c.

References ast_register_cleanup(), stasis_cache_cleanup(), stasis_cache_clear_type(), stasis_cache_update_type(), and STASIS_MESSAGE_TYPE_INIT.

Referenced by stasis_init().

 {
    ast_register_cleanup(stasis_cache_cleanup);
 
    if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_clear_type) != 0) {
       return -1;
    }
 
    if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_update_type) != 0) {
       return -1;
    }
 
    return 0;
 }

◆ stasis_cache_update_dtor()

static void stasis_cache_update_dtor ( void * obj )

static

Definition at line 783 of file stasis_cache.c.

References ao2_cleanup, stasis_cache_update::new_snapshot, NULL, stasis_cache_update::old_snapshot, stasis_cache_update::type, and update().

Referenced by update_create().

 {
    struct stasis_cache_update *update = obj;
 
    ao2_cleanup(update->old_snapshot);
    update->old_snapshot = NULL;
    ao2_cleanup(update->new_snapshot);
    update->new_snapshot = NULL;
    ao2_cleanup(update->type);
    update->type = NULL;
 }

◆ stasis_caching_accept_message_type()

int stasis_caching_accept_message_type	(	struct stasis_caching_topic *	caching_topic,
		struct stasis_message_type *	type
	)

Indicate to a caching topic that we are interested in a message type.

This will cause the caching topic to receive messages of the given message type. This enables internal filtering in the stasis message bus to reduce messages.

Parameters

caching_topic	The caching topic.
type	The message type we wish to receive.

Return values

0	on success
-1	failure

Since: 17.0.0

Definition at line 90 of file stasis_cache.c.

References stasis_cache_clear_type(), stasis_subscription_accept_message_type(), stasis_subscription_change_type(), and stasis_caching_topic::sub.

Referenced by ast_presence_state_engine_init(), devstate_init(), and stasis_cp_single_accept_message_type().

 {
    int res;
 
    if (!caching_topic) {
       return -1;
    }
 
    /* We wait to accept the stasis specific message types until now so that by default everything
     * will flow to us.
     */
    res = stasis_subscription_accept_message_type(caching_topic->sub, stasis_cache_clear_type());
    res |= stasis_subscription_accept_message_type(caching_topic->sub, stasis_subscription_change_type());
    res |= stasis_subscription_accept_message_type(caching_topic->sub, type);
 
    return res;
 }

◆ stasis_caching_get_topic()

struct stasis_topic* stasis_caching_get_topic ( struct stasis_caching_topic * caching_topic )

Returns the topic of cached events from a caching topics.

Parameters

caching_topic The caching topic.

Returns: The topic that publishes cache update events, along with passthrough events from the underlying topic.; NULL if caching_topic is NULL.

Since: 12

Definition at line 85 of file stasis_cache.c.

References stasis_caching_topic::topic.

Referenced by ast_device_state_topic_cached(), ast_mwi_topic_cached(), ast_presence_state_topic_cached(), AST_TEST_DEFINE(), stasis_cp_single_create(), and stasis_cp_single_topic_cached().

 {
    return caching_topic->topic;
 }

◆ stasis_caching_set_filter()

int stasis_caching_set_filter	(	struct stasis_caching_topic *	caching_topic,
		enum stasis_subscription_message_filter	filter
	)

Set the message type filtering level on a cache.

This will cause the underlying subscription to filter messages according to the provided filter level. For example if selective is used then only messages matching those provided to stasis_subscription_accept_message_type will be raised to the subscription callback.

Parameters

caching_topic	The caching topic.
filter	What filter to use

Return values

0	on success
-1	failure

Since: 17.0.0

Definition at line 109 of file stasis_cache.c.

References stasis_subscription_set_filter(), and stasis_caching_topic::sub.

Referenced by ast_presence_state_engine_init(), devstate_init(), and stasis_cp_single_set_filter().

 {
    if (!caching_topic) {
       return -1;
    }
    return stasis_subscription_set_filter(caching_topic->sub, filter);
 }

◆ stasis_caching_topic_create()

struct stasis_caching_topic* stasis_caching_topic_create	(	struct stasis_topic *	original_topic,
		struct stasis_cache *	cache
	)

Create a topic which monitors and caches messages from another topic.

The idea is that some topics publish 'snapshots' of some other object's state that should be cached. When these snapshot messages are received, the cache is updated, and a stasis_cache_update() message is forwarded, which has both the original snapshot message and the new message.

The returned object is AO2 managed, so ao2_cleanup() when done with it.

Parameters

original_topic	Topic publishing snapshot messages.
cache	Backend cache in which to keep snapshots.

Returns: New topic which changes snapshot messages to stasis_cache_update() messages, and forwards all other messages from the original topic.; NULL on error

Since: 12

Definition at line 948 of file stasis_cache.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_container_register(), ao2_ref, ast_asprintf, ast_atomic_fetchadd_int(), ast_free, ast_log, stasis_caching_topic::cache, cache, caching_topic_exec(), stasis_cache::entries, internal_stasis_subscribe(), LOG_ERROR, NULL, stasis_caching_topic::original_topic, print_cache_entry(), stasis_cache::registered, stasis_caching_topic_dtor(), stasis_topic_create(), stasis_topic_name(), stasis_caching_topic::sub, and stasis_caching_topic::topic.

Referenced by ast_presence_state_engine_init(), AST_TEST_DEFINE(), devstate_init(), mwi_init(), and stasis_cp_sink_create().

 {
    struct stasis_caching_topic *caching_topic;
    static int caching_id;
    char *new_name;
    int ret;
 
    ret = ast_asprintf(&new_name, "cache:%d/%s", ast_atomic_fetchadd_int(&caching_id, +1), stasis_topic_name(original_topic));
    if (ret < 0) {
       return NULL;
    }
 
    caching_topic = ao2_alloc_options(sizeof(*caching_topic),
       stasis_caching_topic_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (caching_topic == NULL) {
       ast_free(new_name);
 
       return NULL;
    }
 
    caching_topic->topic = stasis_topic_create(new_name);
    if (caching_topic->topic == NULL) {
       ao2_ref(caching_topic, -1);
       ast_free(new_name);
 
       return NULL;
    }
 
    ao2_ref(cache, +1);
    caching_topic->cache = cache;
    if (!cache->registered) {
       if (ao2_container_register(new_name, cache->entries, print_cache_entry)) {
          ast_log(LOG_ERROR, "Stasis cache container '%p' for '%s' did not register\n",
             cache->entries, new_name);
       } else {
          cache->registered = 1;
       }
    }
    ast_free(new_name);
 
    caching_topic->sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0, 0, __FILE__, __LINE__, __PRETTY_FUNCTION__);
    if (caching_topic->sub == NULL) {
       ao2_ref(caching_topic, -1);
 
       return NULL;
    }
 
    ao2_ref(original_topic, +1);
    caching_topic->original_topic = original_topic;
 
    /* The subscription holds the reference, so no additional ref bump. */
    return caching_topic;
 }

◆ stasis_caching_topic_dtor()

static void stasis_caching_topic_dtor ( void * obj )

static

Definition at line 62 of file stasis_cache.c.

References ao2_cleanup, ao2_container_unregister(), ast_assert, stasis_caching_topic::cache, NULL, stasis_caching_topic::original_topic, stasis_subscription_is_done(), stasis_subscription_is_subscribed(), stasis_topic_name(), stasis_caching_topic::sub, and stasis_caching_topic::topic.

Referenced by stasis_caching_topic_create().

 {
    struct stasis_caching_topic *caching_topic = obj;
 
    /* Caching topics contain subscriptions, and must be manually
     * unsubscribed. */
    ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
    /* If there are any messages in flight to this subscription; that would
     * be bad. */
    ast_assert(stasis_subscription_is_done(caching_topic->sub));
 
    ao2_container_unregister(stasis_topic_name(caching_topic->topic));
 
    ao2_cleanup(caching_topic->sub);
    caching_topic->sub = NULL;
    ao2_cleanup(caching_topic->cache);
    caching_topic->cache = NULL;
    ao2_cleanup(caching_topic->topic);
    caching_topic->topic = NULL;
    ao2_cleanup(caching_topic->original_topic);
    caching_topic->original_topic = NULL;
 }

◆ stasis_caching_unsubscribe()

struct stasis_caching_topic* stasis_caching_unsubscribe ( struct stasis_caching_topic * caching_topic )

Unsubscribes a caching topic from its upstream topic.

This function returns immediately, so be sure to cleanup when stasis_subscription_final_message() is received.

Parameters

caching_topic Caching topic to unsubscribe

Returns: NULL for convenience

Since: 12

Definition at line 119 of file stasis_cache.c.

References ao2_cleanup, ao2_ref, ast_log, LOG_ERROR, NULL, stasis_subscription_is_subscribed(), stasis_unsubscribe(), and stasis_caching_topic::sub.

Referenced by AST_TEST_DEFINE(), stasis_caching_unsubscribe_and_join(), and stasis_cp_single_unsubscribe().

 {
    if (!caching_topic) {
       return NULL;
    }
 
    /*
     * The subscription may hold the last reference to this caching
     * topic, but we want to make sure the unsubscribe finishes
     * before kicking of the caching topic's dtor.
     */
    ao2_ref(caching_topic, +1);
 
    if (stasis_subscription_is_subscribed(caching_topic->sub)) {
       /*
        * Increment the reference to hold on to it past the
        * unsubscribe. Will be cleaned up in dtor.
        */
       ao2_ref(caching_topic->sub, +1);
       stasis_unsubscribe(caching_topic->sub);
    } else {
       ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
    }
    ao2_cleanup(caching_topic);
    return NULL;
 }

◆ stasis_caching_unsubscribe_and_join()

struct stasis_caching_topic* stasis_caching_unsubscribe_and_join ( struct stasis_caching_topic * caching_topic )

Unsubscribes a caching topic from its upstream topic, blocking until all messages have been forwarded.

See stasis_unsubscriben_and_join() for more info on when to use this as opposed to stasis_caching_unsubscribe().

Parameters

caching_topic Caching topic to unsubscribe

Returns: NULL for convenience

Since: 12

Definition at line 146 of file stasis_cache.c.

References ao2_cleanup, ao2_ref, NULL, stasis_caching_unsubscribe(), stasis_subscription_join(), and stasis_caching_topic::sub.

Referenced by AST_TEST_DEFINE(), devstate_cleanup(), mwi_cleanup(), and presence_state_engine_cleanup().

 {
    if (!caching_topic) {
       return NULL;
    }
 
    /* Hold a ref past the unsubscribe */
    ao2_ref(caching_topic, +1);
    stasis_caching_unsubscribe(caching_topic);
    stasis_subscription_join(caching_topic->sub);
    ao2_cleanup(caching_topic);
    return NULL;
 }

◆ STASIS_MESSAGE_TYPE_DEFN() [1/2]

STASIS_MESSAGE_TYPE_DEFN ( stasis_cache_clear_type )

Referenced by stasis_cache_dump_all().

◆ STASIS_MESSAGE_TYPE_DEFN() [2/2]

STASIS_MESSAGE_TYPE_DEFN ( stasis_cache_update_type )

◆ update_create()

static struct stasis_message* update_create	(	struct stasis_message *	old_snapshot,
		struct stasis_message *	new_snapshot
	)

static

Definition at line 795 of file stasis_cache.c.

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_cleanup, ao2_ref, ast_assert, stasis_cache_update::new_snapshot, NULL, stasis_cache_update::old_snapshot, stasis_cache_update_dtor(), stasis_cache_update_type(), stasis_message_create(), stasis_message_type(), stasis_cache_update::type, and update().

Referenced by caching_topic_exec().

 {
    struct stasis_cache_update *update;
    struct stasis_message *msg;
 
    ast_assert(old_snapshot != NULL || new_snapshot != NULL);
 
    if (!stasis_cache_update_type()) {
       return NULL;
    }
 
    update = ao2_alloc_options(sizeof(*update), stasis_cache_update_dtor,
       AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (!update) {
       return NULL;
    }
 
    if (old_snapshot) {
       ao2_ref(old_snapshot, +1);
       update->old_snapshot = old_snapshot;
       if (!new_snapshot) {
          ao2_ref(stasis_message_type(old_snapshot), +1);
          update->type = stasis_message_type(old_snapshot);
       }
    }
    if (new_snapshot) {
       ao2_ref(new_snapshot, +1);
       update->new_snapshot = new_snapshot;
       ao2_ref(stasis_message_type(new_snapshot), +1);
       update->type = stasis_message_type(new_snapshot);
    }
 
    msg = stasis_message_create(stasis_cache_update_type(), update);
 
    ao2_cleanup(update);
    return msg;
 }

Data Structures

Macros

Functions

Detailed Description

Macro Definition Documentation

◆ NUM_CACHE_BUCKETS

Function Documentation

◆ cache_dtor()

◆ cache_dump_all_cb()

◆ cache_dump_by_eid_cb()

◆ cache_entry_by_eid()

◆ cache_entry_cmp()

◆ cache_entry_compute_hash()

◆ cache_entry_create()

◆ cache_entry_dtor()

◆ cache_entry_dump()

◆ cache_entry_hash()

◆ cache_find()

◆ cache_put()

◆ cache_remove()

◆ cache_udpate()

◆ caching_topic_exec()

◆ print_cache_entry()

◆ stasis_cache_cleanup()

◆ stasis_cache_clear_create()

◆ stasis_cache_create()

◆ stasis_cache_create_full()

◆ stasis_cache_dump()

◆ stasis_cache_dump_all()

◆ stasis_cache_dump_by_eid()

◆ stasis_cache_entry_get_aggregate()

◆ stasis_cache_entry_get_local()

◆ stasis_cache_entry_get_remote()

◆ stasis_cache_get()

◆ stasis_cache_get_all()

◆ stasis_cache_get_by_eid()

◆ stasis_cache_init()

◆ stasis_cache_update_dtor()

◆ stasis_caching_accept_message_type()

◆ stasis_caching_get_topic()

◆ stasis_caching_set_filter()

◆ stasis_caching_topic_create()

◆ stasis_caching_topic_dtor()

◆ stasis_caching_unsubscribe()

◆ stasis_caching_unsubscribe_and_join()

◆ STASIS_MESSAGE_TYPE_DEFN() [1/2]

◆ STASIS_MESSAGE_TYPE_DEFN() [2/2]

◆ update_create()