Asterisk - The Open Source Telephony Project
21.4.1
|
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"
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) |
Stasis Message API.
Definition in file stasis_cache.c.
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.
message | Message representative of the cache entry that should be cleared. This will become the data held in the stasis_cache_clear message. |
NULL | on error. |
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(), and ast_endpoint_shutdown().
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.
id_fn | Callback to extract the id from a snapshot message. |
NULL | on error |
Definition at line 360 of file stasis_cache.c.
References stasis_cache_create_full().
Referenced by mwi_init(), and stasis_cp_all_create().
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.
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. |
NULL | on error |
Definition at line 334 of file stasis_cache.c.
References AO2_ALLOC_OPT_LOCK_NOLOCK, AO2_ALLOC_OPT_LOCK_RWLOCK, ao2_container_alloc_hash, and cache.
Referenced by devstate_init(), and stasis_cache_create().
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.
cache | The cache to query. |
type | Type of message to dump (any type if NULL ). |
NULL | on allocation error |
Definition at line 736 of file stasis_cache.c.
References ast_eid_default, and stasis_cache_dump_by_eid().
Referenced by ast_ari_endpoints_list(), ast_ari_endpoints_list_by_tech(), and xmpp_init_event_distribution().
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.
cache | The cache to query. |
type | Type of message to dump (any type if NULL ). |
NULL | on allocation error |
Definition at line 757 of file stasis_cache.c.
References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, OBJ_MULTIPLE, and OBJ_NODATA.
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.
cache | The cache to query. |
type | Type of message to dump (any type if NULL ). |
eid | Specific entity id to retrieve. NULL for aggregate. |
NULL | on allocation error |
Definition at line 718 of file stasis_cache.c.
References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_callback, ao2_container_alloc_list, OBJ_MULTIPLE, and OBJ_NODATA.
Referenced by stasis_cache_dump().
struct stasis_message* stasis_cache_entry_get_aggregate | ( | struct stasis_cache_entry * | entry | ) |
Get the aggregate cache entry snapshot.
entry | Cache entry to get the aggregate snapshot. |
Aggregate-snapshot | in cache. |
NULL | if not present. |
Definition at line 365 of file stasis_cache.c.
References stasis_cache_entry::aggregate.
struct stasis_message* stasis_cache_entry_get_local | ( | struct stasis_cache_entry * | entry | ) |
Get the local entity's cache entry snapshot.
entry | Cache entry to get the local entity's snapshot. |
Internal-snapshot | in cache. |
NULL | if not present. |
Definition at line 370 of file stasis_cache.c.
References stasis_cache_entry::local.
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.
entry | Cache entry to get a remote entity's snapshot. |
idx | Which remote entity's snapshot to get. |
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 stasis_cache_entry::remote.
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.
cache | The cache to query. |
type | Type of message to retrieve. |
id | Identity of the snapshot to retrieve. |
NULL | if message is not found. |
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(), and update_registry().
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.
cache | The cache to query. |
type | Type of message to retrieve. |
id | Identity of the snapshot to retrieve. |
NULL | if error. |
Definition at line 587 of file stasis_cache.c.
References AO2_ALLOC_OPT_LOCK_NOLOCK, and ao2_container_alloc_list.
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.
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. |
NULL | if message is not found. |
Definition at line 659 of file stasis_cache.c.
References ao2_bump.
Referenced by ast_delete_mwi_state_full(), ast_device_state_clear_cache(), and stasis_cache_get().
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.
caching_topic | The caching topic. |
type | The message type we wish to receive. |
0 | on success |
-1 | failure |
Definition at line 90 of file stasis_cache.c.
References stasis_cache_clear_type(), stasis_subscription_accept_message_type(), and stasis_subscription_change_type().
Referenced by devstate_init(), and stasis_cp_single_accept_message_type().
struct stasis_topic* stasis_caching_get_topic | ( | struct stasis_caching_topic * | caching_topic | ) |
Returns the topic of cached events from a caching topics.
caching_topic | The caching topic. |
NULL | if caching_topic is NULL . |
Definition at line 85 of file stasis_cache.c.
Referenced by ast_device_state_topic_cached(), ast_mwi_topic_cached(), ast_presence_state_topic_cached(), stasis_cp_single_create(), and stasis_cp_single_topic_cached().
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.
caching_topic | The caching topic. |
filter | What filter to use |
0 | on success |
-1 | failure |
Definition at line 109 of file stasis_cache.c.
References stasis_subscription_set_filter().
Referenced by devstate_init(), and stasis_cp_single_set_filter().
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.
original_topic | Topic publishing snapshot messages. |
cache | Backend cache in which to keep snapshots. |
NULL | on error |
Definition at line 948 of file stasis_cache.c.
References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_container_register(), ao2_ref, ast_asprintf, ast_atomic_fetchadd_int(), cache, internal_stasis_subscribe(), stasis_topic_create(), and stasis_topic_name().
Referenced by devstate_init(), mwi_init(), and stasis_cp_sink_create().
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.
caching_topic | Caching topic to unsubscribe |
NULL | for convenience |
Definition at line 119 of file stasis_cache.c.
References ao2_ref, stasis_subscription_is_subscribed(), and stasis_unsubscribe().
Referenced by stasis_caching_unsubscribe_and_join(), and stasis_cp_single_unsubscribe().
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().
caching_topic | Caching topic to unsubscribe |
NULL | for convenience |
Definition at line 146 of file stasis_cache.c.
References ao2_ref, stasis_caching_unsubscribe(), and stasis_subscription_join().