stasis_cache.c File Reference

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_update (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.

Function Documentation

cache_dtor()

static void cache_dtor ( void *  obj )

static

Definition at line 326 of file stasis_cache.c.

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

References ao2_cleanup, cache, and NULL.

Referenced by stasis_cache_create_full().

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.

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

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

Referenced by stasis_cache_dump_all().

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.

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

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

Referenced by stasis_cache_dump_by_eid().

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.

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

References ast_eid_cmp(), ast_eid_default, AST_VECTOR_GET, AST_VECTOR_SIZE, stasis_message::eid, NULL, and stasis_message_eid().

Referenced by cache_dump_by_eid_cb(), and stasis_cache_get_by_eid().

cache_entry_cmp()

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

static

Definition at line 288 of file stasis_cache.c.

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

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().

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.

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

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().

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.

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

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(), entry::key, NULL, stasis_message_eid(), stasis_message_type(), and type.

Referenced by cache_put().

cache_entry_dtor()

static void cache_entry_dtor ( void *  obj )

static

Definition at line 183 of file stasis_cache.c.

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

References ao2_cleanup, ast_free, AST_VECTOR_FREE, AST_VECTOR_GET, AST_VECTOR_SIZE, entry::key, and NULL.

Referenced by cache_entry_create().

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.

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

References ao2_link, ast_assert, AST_VECTOR_GET, AST_VECTOR_SIZE, and NULL.

Referenced by cache_dump_all_cb(), and stasis_cache_get_all().

cache_entry_hash()

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

static

Definition at line 266 of file stasis_cache.c.

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

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

Referenced by stasis_cache_create_full().

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.

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

References ao2_find, ast_assert, cache_entry_compute_hash(), id, cache_entry_key::id, 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().

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.

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

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, cache_entry_create(), cache_find(), cache_remove(), cache_update(), NULL, OBJ_NOLOCK, cache_put_snapshots::old, stasis_message_type(), and type.

Referenced by caching_topic_exec().

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.

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

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

Referenced by cache_put(), and caching_topic_exec().

cache_update()

static struct stasis_message * cache_update ( 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.

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

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

Referenced by cache_put(), and consumer_exec().

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.

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

References cache_put_snapshots::aggregate_new, cache_put_snapshots::aggregate_old, 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, cache_put_snapshots::old, 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(), sub, stasis_caching_topic::topic, stasis_subscription_change::uniqueid, update(), and update_create().

Referenced by stasis_caching_topic_create().

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.

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

References entry::key, and stasis_message_type_name().

Referenced by stasis_caching_topic_create().

stasis_cache_cleanup()

static void stasis_cache_cleanup ( void  )

static

Definition at line 1002 of file stasis_cache.c.

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

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

Referenced by stasis_cache_init().

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.

Return values

NULL

on error.

Since

12

Definition at line 778 of file stasis_cache.c.

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

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().

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.

Returns

New cache indexed by id_fn.

Return values

NULL

on error

Since

12

Definition at line 360 of file stasis_cache.c.

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

References stasis_cache::id_fn, NULL, and stasis_cache_create_full().

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

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.

Returns

New cache indexed by id_fn.

Return values

NULL

on error

Since

12.2.0

Definition at line 334 of file stasis_cache.c.

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

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::id_fn, NULL, and NUM_CACHE_BUCKETS.

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

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).

Returns

ao2_container containing all matches (must be unreffed by caller)

Return values

NULL	on allocation error

Since

12

Definition at line 736 of file stasis_cache.c.

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

References ast_eid_default, cache, stasis_cache_dump_by_eid(), and type.

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().

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).

Returns

ao2_container containing all matches (must be unreffed by caller)

Return values

NULL	on allocation error

Definition at line 757 of file stasis_cache.c.

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

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, ast_assert, cache, cache_dump_all_cb(), cache_dump_data::container, cache_dump_data::eid, NULL, OBJ_MULTIPLE, OBJ_NODATA, type, and cache_dump_data::type.

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

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.

Returns

ao2_container containing all matches (must be unreffed by caller)

Return values

NULL	on allocation error

Definition at line 718 of file stasis_cache.c.

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

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, ast_assert, cache, cache_dump_by_eid_cb(), cache_dump_data::container, cache_dump_data::eid, 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().

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.

{
    return entry->aggregate;
}

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

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.

{
    return entry->local;
}

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

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.

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

References AST_VECTOR_GET, AST_VECTOR_SIZE, and NULL.

Referenced by cache_test_aggregate_calc_fn(), and device_state_aggregate_calc().

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.

Returns

Message from the cache.

Return values

NULL	if message is not found.

Since

12

Definition at line 686 of file stasis_cache.c.

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

References ast_eid_default, cache, stasis_cache_get_by_eid(), and type.

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

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.

Returns

Container of matching items found.

Return values

NULL

if error.

Definition at line 587 of file stasis_cache.c.

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

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_cleanup, ao2_container_alloc_list, ao2_rdlock, ao2_unlock, ast_assert, cache, cache_entry_dump(), cache_find(), NULL, and type.

Referenced by AST_TEST_DEFINE().

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.

Returns

Message from the cache.

Return values

NULL	if message is not found.

Since

12.2.0

Definition at line 659 of file stasis_cache.c.

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

References ao2_bump, ao2_cleanup, ao2_rdlock, ao2_unlock, ast_assert, cache, cache_entry_by_eid(), cache_find(), stasis_message::eid, NULL, and type.

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

stasis_cache_init()

int stasis_cache_init

(

void

)

Definition at line 1008 of file stasis_cache.c.

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

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

Referenced by stasis_init().

stasis_cache_update_dtor()

static void stasis_cache_update_dtor ( void *  obj )

static

Definition at line 783 of file stasis_cache.c.

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

References ao2_cleanup, NULL, and update().

Referenced by update_create().

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.

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

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

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

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.

Return values

NULL	if caching_topic is NULL.

Since

12

Definition at line 85 of file stasis_cache.c.

{
    return caching_topic->topic;
}

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().

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.

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

References filter(), stasis_subscription_set_filter(), and stasis_caching_topic::sub.

Referenced by ast_presence_state_engine_init(), devstate_init(), and stasis_cp_single_set_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.

Return values

NULL

on error

Since

12

Definition at line 948 of file stasis_cache.c.

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

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(), internal_stasis_subscribe(), LOG_ERROR, NULL, stasis_caching_topic::original_topic, print_cache_entry(), 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().

stasis_caching_topic_dtor()

static void stasis_caching_topic_dtor ( void *  obj )

static

Definition at line 62 of file stasis_cache.c.

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

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().

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

Return values

NULL	for convenience

Since

12

Definition at line 119 of file stasis_cache.c.

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

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().

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_unsubscribe_and_join() for more info on when to use this as opposed to stasis_caching_unsubscribe().

Parameters

caching_topic

Caching topic to unsubscribe

Return values

NULL	for convenience

Since

12

Definition at line 146 of file stasis_cache.c.

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

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().

STASIS_MESSAGE_TYPE_DEFN() [1/2]

STASIS_MESSAGE_TYPE_DEFN

(

stasis_cache_clear_type

)

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.

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

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_cleanup, ao2_ref, ast_assert, NULL, stasis_cache_update_dtor(), stasis_cache_update_type(), stasis_message_create(), stasis_message_type(), and update().

Referenced by caching_topic_exec().