stasis_cache_pattern.c File Reference

Typical cache pattern for Stasis topics. More...

#include "asterisk.h"
#include "asterisk/astobj2.h"
#include "asterisk/stasis_cache_pattern.h"

Include dependency graph for stasis_cache_pattern.c:

Go to the source code of this file.

Data Structures
struct	stasis_cp_all
struct	stasis_cp_single

Functions
static void	all_dtor (void *obj)
static void	one_dtor (void *obj)
struct stasis_cache *	stasis_cp_all_cache (struct stasis_cp_all *all)
	Get the cache. More...
struct stasis_cp_all *	stasis_cp_all_create (const char *name, snapshot_get_id id_fn)
	Create an all instance of the cache pattern. More...
struct stasis_topic *	stasis_cp_all_topic (struct stasis_cp_all *all)
	Get the aggregate topic. More...
struct stasis_topic *	stasis_cp_all_topic_cached (struct stasis_cp_all *all)
	Get the caching topic. More...
int	stasis_cp_single_accept_message_type (struct stasis_cp_single *one, struct stasis_message_type *type)
	Indicate to an instance that we are interested in a message type. More...
struct stasis_cp_single *	stasis_cp_single_create (struct stasis_cp_all *all, const char *name)
	Create the 'one' side of the cache pattern. More...
int	stasis_cp_single_set_filter (struct stasis_cp_single *one, enum stasis_subscription_message_filter filter)
	Set the message type filtering level on a cache. More...
struct stasis_topic *	stasis_cp_single_topic (struct stasis_cp_single *one)
	Get the topic for this instance. More...
struct stasis_topic *	stasis_cp_single_topic_cached (struct stasis_cp_single *one)
	Get the caching topic for this instance. More...
void	stasis_cp_single_unsubscribe (struct stasis_cp_single *one)
	Stops caching and forwarding messages. More...
struct stasis_cp_single *	stasis_cp_sink_create (struct stasis_cp_all *all, const char *name)
	Create a sink in the cache pattern. More...

Detailed Description

Typical cache pattern for Stasis topics.

Author

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

Definition in file stasis_cache_pattern.c.

Function Documentation

all_dtor()

static void all_dtor ( void *  obj )

static

Definition at line 51 of file stasis_cache_pattern.c.

References ao2_cleanup, stasis_cp_all::cache, stasis_cp_all::forward_all_to_cached, NULL, stasis_forward_cancel(), stasis_cp_all::topic, and stasis_cp_all::topic_cached.

Referenced by stasis_cp_all_create().

{
   struct stasis_cp_all *all = obj;

   ao2_cleanup(all->topic);
   all->topic = NULL;
   ao2_cleanup(all->topic_cached);
   all->topic_cached = NULL;
   ao2_cleanup(all->cache);
   all->cache = NULL;
   stasis_forward_cancel(all->forward_all_to_cached);
   all->forward_all_to_cached = NULL;
}

one_dtor()

static void one_dtor ( void *  obj )

static

Definition at line 126 of file stasis_cache_pattern.c.

References ao2_cleanup, ast_assert, stasis_cp_single::forward_cached_to_all, stasis_cp_single::forward_topic_to_all, NULL, stasis_cp_single::topic, and stasis_cp_single::topic_cached.

Referenced by stasis_cp_sink_create().

{
   struct stasis_cp_single *one = obj;

   /* Should already be unsubscribed */
   ast_assert(one->topic_cached == NULL);
   ast_assert(one->forward_topic_to_all == NULL);
   ast_assert(one->forward_cached_to_all == NULL);

   ao2_cleanup(one->topic);
   one->topic = NULL;
}

stasis_cp_all_cache()

struct stasis_cache* stasis_cp_all_cache

(

struct stasis_cp_all *

all

)

Get the cache.

This is the shared cache for all corresponding stasis_cp_single objects.

Parameters

all	All side caching pattern object.

Returns

The cache.

NULL if all is NULL

Definition at line 118 of file stasis_cache_pattern.c.

References stasis_cp_all::cache, and NULL.

Referenced by ast_endpoint_cache().

{
   if (!all) {
      return NULL;
   }
   return all->cache;
}

stasis_cp_all_create()

struct stasis_cp_all* stasis_cp_all_create	(	const char *	name,
		snapshot_get_id	id_fn
	)

Create an all instance of the cache pattern.

This object is AO2 managed, so dispose of it with ao2_cleanup().

Parameters

name	Base name of the topics.
id_fn	Identity function for the cache.

Returns

All side instance.

NULL on error.

Definition at line 65 of file stasis_cache_pattern.c.

References all_dtor(), ao2_ref, ao2_t_alloc, ast_asprintf, ast_atomic_fetchadd_int(), ast_free, stasis_cp_all::cache, stasis_cp_all::forward_all_to_cached, NULL, stasis_cache_create(), stasis_forward_all(), stasis_topic_create(), stasis_cp_all::topic, and stasis_cp_all::topic_cached.

Referenced by ast_endpoint_stasis_init().

{
   char *cached_name = NULL;
   struct stasis_cp_all *all;
   static int cache_id;

   all = ao2_t_alloc(sizeof(*all), all_dtor, name);
   if (!all) {
      return NULL;
   }

   ast_asprintf(&cached_name, "cache_pattern:%d/%s", ast_atomic_fetchadd_int(&cache_id, +1), name);
   if (!cached_name) {
      ao2_ref(all, -1);

      return NULL;
   }

   all->topic = stasis_topic_create(name);
   all->topic_cached = stasis_topic_create(cached_name);
   ast_free(cached_name);
   all->cache = stasis_cache_create(id_fn);
   all->forward_all_to_cached =
      stasis_forward_all(all->topic, all->topic_cached);

   if (!all->topic || !all->topic_cached || !all->cache ||
      !all->forward_all_to_cached) {
      ao2_ref(all, -1);

      return NULL;
   }

   return all;
}

stasis_cp_all_topic()

struct stasis_topic* stasis_cp_all_topic

(

struct stasis_cp_all *

all

)

Get the aggregate topic.

This topic aggregates all messages published to corresponding stasis_cp_single_topic() topics.

Parameters

all	All side caching pattern object.

Returns

The aggregate topic.

NULL if all is NULL

Definition at line 101 of file stasis_cache_pattern.c.

References NULL, and stasis_cp_all::topic.

Referenced by ast_endpoint_topic_all().

{
   if (!all) {
      return NULL;
   }
   return all->topic;
}

stasis_cp_all_topic_cached()

struct stasis_topic* stasis_cp_all_topic_cached

(

struct stasis_cp_all *

all

)

Get the caching topic.

This topic aggregates all messages from the corresponding stasis_cp_single_topic_cached() topics.

Note that one normally only subscribes to the caching topic, since data is fed to it from its upstream topic.

Parameters

all	All side caching pattern object.

Returns

The aggregate caching topic.

NULL if all is NULL

Definition at line 109 of file stasis_cache_pattern.c.

References NULL, and stasis_cp_all::topic_cached.

Referenced by ast_endpoint_topic_all_cached().

{
   if (!all) {
      return NULL;
   }
   return all->topic_cached;
}

stasis_cp_single_accept_message_type()

int stasis_cp_single_accept_message_type	(	struct stasis_cp_single *	one,
		struct stasis_message_type *	type
	)

Indicate to an instance 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

one	One side of the cache pattern.
type	The message type we wish to receive.

Return values

0	on success
-1	failure

Since

17.0.0

Definition at line 222 of file stasis_cache_pattern.c.

References stasis_caching_accept_message_type(), and stasis_cp_single::topic_cached.

Referenced by endpoint_internal_create().

{
   if (!one) {
      return -1;
   }
   return stasis_caching_accept_message_type(one->topic_cached, type);
}

stasis_cp_single_create()

struct stasis_cp_single* stasis_cp_single_create	(	struct stasis_cp_all *	all,
		const char *	name
	)

Create the 'one' side of the cache pattern.

Create the 'one' and forward to all's topic and topic_cached.

Dispose of using stasis_cp_single_unsubscribe().

Parameters

all	Corresponding all side.
name	Base name for the topics.

Returns

One side instance

Definition at line 139 of file stasis_cache_pattern.c.

References ao2_ref, stasis_cp_single::forward_cached_to_all, stasis_cp_single::forward_topic_to_all, NULL, stasis_caching_get_topic(), stasis_cp_sink_create(), stasis_forward_all(), stasis_cp_all::topic, stasis_cp_single::topic, stasis_cp_all::topic_cached, and stasis_cp_single::topic_cached.

Referenced by endpoint_internal_create().

{
   struct stasis_cp_single *one;

   one = stasis_cp_sink_create(all, name);
   if (!one) {
      return NULL;
   }

   one->forward_topic_to_all = stasis_forward_all(one->topic, all->topic);
   one->forward_cached_to_all = stasis_forward_all(
      stasis_caching_get_topic(one->topic_cached), all->topic_cached);

   if (!one->forward_topic_to_all || !one->forward_cached_to_all) {
      ao2_ref(one, -1);

      return NULL;
   }

   return one;
}

stasis_cp_single_set_filter()

int stasis_cp_single_set_filter	(	struct stasis_cp_single *	one,
		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

one	One side of the cache pattern.
filter	What filter to use

Return values

0	on success
-1	failure

Since

17.0.0

Definition at line 231 of file stasis_cache_pattern.c.

References stasis_caching_set_filter(), and stasis_cp_single::topic_cached.

Referenced by endpoint_internal_create().

{
   if (!one) {
      return -1;
   }
   return stasis_caching_set_filter(one->topic_cached, filter);
}

stasis_cp_single_topic()

struct stasis_topic* stasis_cp_single_topic

(

struct stasis_cp_single *

one

)

Get the topic for this instance.

This is the topic to which one would post instance-specific messages, or subscribe for single-instance, uncached messages.

Parameters

one	One side of the cache pattern.

Returns

The main topic.

NULL if one is NULL

Definition at line 205 of file stasis_cache_pattern.c.

References NULL, and stasis_cp_single::topic.

Referenced by ast_endpoint_topic(), and endpoint_internal_create().

{
   if (!one) {
      return NULL;
   }
   return one->topic;
}

stasis_cp_single_topic_cached()

struct stasis_topic* stasis_cp_single_topic_cached

(

struct stasis_cp_single *

one

)

Get the caching topic for this instance.

Note that one normally only subscribes to the caching topic, since data is fed to it from its upstream topic.

Parameters

one	One side of the cache pattern.

Returns

The caching topic.

NULL if one is NULL

Definition at line 213 of file stasis_cache_pattern.c.

References NULL, stasis_caching_get_topic(), and stasis_cp_single::topic_cached.

Referenced by ast_endpoint_topic_cached().

{
   if (!one) {
      return NULL;
   }
   return stasis_caching_get_topic(one->topic_cached);
}

stasis_cp_single_unsubscribe()

void stasis_cp_single_unsubscribe

(

struct stasis_cp_single *

one

)

Stops caching and forwarding messages.

Parameters

one	One side of the cache pattern.

Definition at line 189 of file stasis_cache_pattern.c.

References ao2_cleanup, stasis_cp_single::forward_cached_to_all, stasis_cp_single::forward_topic_to_all, NULL, stasis_caching_unsubscribe(), stasis_forward_cancel(), and stasis_cp_single::topic_cached.

Referenced by endpoint_dtor().

{
   if (!one) {
      return;
   }

   stasis_forward_cancel(one->forward_topic_to_all);
   one->forward_topic_to_all = NULL;
   stasis_forward_cancel(one->forward_cached_to_all);
   one->forward_cached_to_all = NULL;
   stasis_caching_unsubscribe(one->topic_cached);
   one->topic_cached = NULL;

   ao2_cleanup(one);
}

stasis_cp_sink_create()

struct stasis_cp_single* stasis_cp_sink_create	(	struct stasis_cp_all *	all,
		const char *	name
	)

Create a sink in the cache pattern.

Create the 'one' but do not automatically forward to the all's topic. This is useful when aggregating other topic's messages created with stasis_cp_single_create in another caching topic without replicating those messages in the all's topics.

Dispose of using stasis_cp_single_unsubscribe().

Parameters

all	Corresponding all side.
name	Base name for the topics.

Returns

One side instance

Definition at line 162 of file stasis_cache_pattern.c.

References ao2_ref, ao2_t_alloc, stasis_cp_all::cache, NULL, one_dtor(), stasis_caching_topic_create(), stasis_topic_create(), stasis_cp_single::topic, and stasis_cp_single::topic_cached.

Referenced by endpoint_internal_create(), and stasis_cp_single_create().

{
   struct stasis_cp_single *one;

   one = ao2_t_alloc(sizeof(*one), one_dtor, name);
   if (!one) {
      return NULL;
   }

   one->topic = stasis_topic_create(name);
   if (!one->topic) {
      ao2_ref(one, -1);

      return NULL;
   }

   one->topic_cached = stasis_caching_topic_create(one->topic, all->cache);
   if (!one->topic_cached) {
      ao2_ref(one, -1);

      return NULL;
   }

   return one;
}