Asterisk - The Open Source Telephony Project  21.4.1
Data Structures | Macros | Functions | Variables
messaging.c File Reference

Stasis out-of-call text message support. More...

#include "asterisk.h"
#include "asterisk/message.h"
#include "asterisk/endpoints.h"
#include "asterisk/astobj2.h"
#include "asterisk/vector.h"
#include "asterisk/lock.h"
#include "asterisk/utils.h"
#include "asterisk/test.h"
#include "messaging.h"

Go to the source code of this file.

Data Structures

struct  application_tuple
 Storage object for an application. More...
 
struct  message_subscription
 A subscription to some endpoint or technology. More...
 

Macros

#define ENDPOINTS_NUM_BUCKETS   127
 Number of buckets for the endpoint_subscriptions container.
 
#define TECH_WILDCARD   "__AST_ALL_TECH"
 Subscription to all technologies.
 

Functions

static struct application_tupleapplication_tuple_alloc (const char *app_name, message_received_cb callback, void *pvt)
 
static int application_tuple_cmp (struct application_tuple *item, const char *key)
 
static void application_tuple_dtor (void *obj)
 
static void dispatch_message (struct message_subscription *sub, const char *endpoint_name, struct ast_json *json_msg)
 
static struct message_subscriptionget_or_create_subscription (struct ast_endpoint *endpoint)
 
static struct message_subscriptionget_subscription (struct ast_endpoint *endpoint)
 
static int handle_msg_cb (struct ast_msg *msg)
 
static int has_destination_cb (const struct ast_msg *msg)
 
static int is_app_subscribed (struct message_subscription *sub, const char *app_name)
 
static struct message_subscriptionmessage_subscription_alloc (const char *token)
 
static int message_subscription_compare_cb (void *obj, void *arg, int flags)
 
static void message_subscription_dtor (void *obj)
 
static int message_subscription_hash_cb (const void *obj, const int flags)
 
int messaging_app_subscribe_endpoint (const char *app_name, struct ast_endpoint *endpoint, message_received_cb callback, void *pvt)
 Subscribe an application to an endpoint for messages. More...
 
void messaging_app_unsubscribe_endpoint (const char *app_name, const char *endpoint_id)
 Subscribe for messages from a particular endpoint. More...
 
int messaging_cleanup (void)
 Tidy up the messaging layer. More...
 
int messaging_init (void)
 Initialize the messaging layer. More...
 
static int messaging_subscription_cmp (struct message_subscription *sub, const char *key)
 
static void msg_to_endpoint (const struct ast_msg *msg, char *buf, size_t len)
 
static struct ast_jsonmsg_to_json (struct ast_msg *msg)
 

Variables

struct ast_msg_handler ari_msg_handler
 
static struct ao2_containerendpoint_subscriptions
 The subscriptions to endpoints.
 
struct {
   size_t   current
 
   struct message_subscription **   elems
 
   size_t   max
 
tech_subscriptions
 The subscriptions to technologies. More...
 
static ast_rwlock_t tech_subscriptions_lock
 RWLock for tech_subscriptions.
 

Detailed Description

Stasis out-of-call text message support.

Author
Matt Jordan mjord.nosp@m.an@d.nosp@m.igium.nosp@m..com

Definition in file messaging.c.

Function Documentation

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

AO2 comparison function for message_subscription

Definition at line 160 of file messaging.c.

References CMP_MATCH, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, OBJ_SEARCH_PARTIAL_KEY, and message_subscription::token.

Referenced by messaging_init().

161 {
162  const struct message_subscription *object_left = obj;
163  const struct message_subscription *object_right = arg;
164  const char *right_key = arg;
165  int cmp;
166 
167  switch (flags & OBJ_SEARCH_MASK) {
168  case OBJ_SEARCH_OBJECT:
169  right_key = object_right->token;
170  /* Fall through */
171  case OBJ_SEARCH_KEY:
172  cmp = strcmp(object_left->token, right_key);
173  break;
175  /*
176  * We could also use a partial key struct containing a length
177  * so strlen() does not get called for every comparison instead.
178  */
179  cmp = strncmp(object_left->token, right_key, strlen(right_key));
180  break;
181  default:
182  /*
183  * What arg points to is specific to this traversal callback
184  * and has no special meaning to astobj2.
185  */
186  cmp = 0;
187  break;
188  }
189  if (cmp) {
190  return 0;
191  }
192  /*
193  * At this point the traversal callback is identical to a sorted
194  * container.
195  */
196  return CMP_MATCH;
197 }
The arg parameter is a search key, but is not an object.
Definition: astobj2.h:1101
The arg parameter is a partial search key similar to OBJ_SEARCH_KEY.
Definition: astobj2.h:1116
A subscription to some endpoint or technology.
Definition: messaging.c:59
The arg parameter is an object of the same type.
Definition: astobj2.h:1087
Search option field mask.
Definition: astobj2.h:1072
static int message_subscription_hash_cb ( const void *  obj,
const int  flags 
)
static

AO2 hash function for message_subscription

Definition at line 138 of file messaging.c.

References ast_str_hash(), OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, and message_subscription::token.

Referenced by messaging_init().

139 {
140  const struct message_subscription *sub;
141  const char *key;
142 
143  switch (flags & OBJ_SEARCH_MASK) {
144  case OBJ_SEARCH_KEY:
145  key = obj;
146  break;
147  case OBJ_SEARCH_OBJECT:
148  sub = obj;
149  key = sub->token;
150  break;
151  default:
152  /* Hash can only work on something with a full key. */
153  ast_assert(0);
154  return 0;
155  }
156  return ast_str_hash(key);
157 }
The arg parameter is a search key, but is not an object.
Definition: astobj2.h:1101
A subscription to some endpoint or technology.
Definition: messaging.c:59
The arg parameter is an object of the same type.
Definition: astobj2.h:1087
Search option field mask.
Definition: astobj2.h:1072
static force_inline int attribute_pure ast_str_hash(const char *str)
Compute a hash value on a string.
Definition: strings.h:1259
int messaging_app_subscribe_endpoint ( const char *  app_name,
struct ast_endpoint endpoint,
message_received_cb  callback,
void *  pvt 
)

Subscribe an application to an endpoint for messages.

Parameters
app_nameThe name of the Stasis Message Bus API application to subscribe to endpoint
endpointThe endpoint object to subscribe to
callbackThe callback to call when a message is received
pvtAn ao2 ref counted object that will be passed to the callback.
Return values
0subscription was successful
-1subscription failed

Definition at line 493 of file messaging.c.

References ao2_ref, message_subscription::applications, ast_debug, ast_endpoint_get_id(), ast_test_suite_event_notify, AST_VECTOR_APPEND, and RAII_VAR.

Referenced by app_subscribe_endpoint().

494 {
495  RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
496  struct application_tuple *tuple;
497 
498  sub = get_or_create_subscription(endpoint);
499  if (!sub) {
500  return -1;
501  }
502 
503  ao2_lock(sub);
504  if (is_app_subscribed(sub, app_name)) {
505  ao2_unlock(sub);
506  return 0;
507  }
508 
509  tuple = application_tuple_alloc(app_name, callback, pvt);
510  if (!tuple) {
511  ao2_unlock(sub);
512  return -1;
513  }
514  if (AST_VECTOR_APPEND(&sub->applications, tuple)) {
515  ao2_ref(tuple, -1);
516  ao2_unlock(sub);
517  return -1;
518  }
519  ao2_unlock(sub);
520 
521  ast_debug(3, "App '%s' subscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
522  ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Subscribed\r\nAppName: %s\r\nToken: %s\r\n",
523  app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
524 
525  return 0;
526 }
message_received_cb callback
Definition: messaging.c:53
#define AST_VECTOR_APPEND(vec, elem)
Append an element to a vector, growing the vector if needed.
Definition: vector.h:256
Storage object for an application.
Definition: messaging.c:49
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define ast_debug(level,...)
Log a DEBUG message.
#define ast_test_suite_event_notify(s, f,...)
Definition: test.h:189
const char * app_name(struct ast_app *app)
Definition: pbx_app.c:463
A subscription to some endpoint or technology.
Definition: messaging.c:59
#define RAII_VAR(vartype, varname, initval, dtor)
Declare a variable that will call a destructor function when it goes out of scope.
Definition: utils.h:941
const char * ast_endpoint_get_id(const struct ast_endpoint *endpoint)
Gets the tech/resource id of the given endpoint.
void messaging_app_unsubscribe_endpoint ( const char *  app_name,
const char *  endpoint_id 
)

Subscribe for messages from a particular endpoint.

Parameters
app_nameName of the stasis application to unsubscribe from messaging
endpoint_idThe ID of the endpoint we no longer care about

Definition at line 423 of file messaging.c.

References ao2_ref, ao2_unlink, message_subscription::applications, ast_debug, ast_endpoint_find_by_id(), ast_endpoint_get_id(), ast_endpoint_get_resource(), ast_test_suite_event_notify, AST_VECTOR_ELEM_CLEANUP_NOOP, AST_VECTOR_REMOVE_CMP_UNORDERED, AST_VECTOR_SIZE, RAII_VAR, tech_subscriptions, tech_subscriptions_lock, and TECH_WILDCARD.

424 {
425  RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
426  RAII_VAR(struct ast_endpoint *, endpoint, NULL, ao2_cleanup);
427 
428  endpoint = ast_endpoint_find_by_id(endpoint_id);
429  sub = get_subscription(endpoint);
430  if (!sub) {
431  return;
432  }
433 
434  ao2_lock(sub);
435  if (!is_app_subscribed(sub, app_name)) {
436  ao2_unlock(sub);
437  return;
438  }
439 
440  AST_VECTOR_REMOVE_CMP_UNORDERED(&sub->applications, app_name, application_tuple_cmp, ao2_cleanup);
441  if (AST_VECTOR_SIZE(&sub->applications) == 0) {
442  if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
444  } else {
445  ast_rwlock_wrlock(&tech_subscriptions_lock);
447  messaging_subscription_cmp, AST_VECTOR_ELEM_CLEANUP_NOOP);
448  ast_rwlock_unlock(&tech_subscriptions_lock);
449  ao2_ref(sub, -1); /* Release the reference held by tech_subscriptions */
450  }
451  }
452  ao2_unlock(sub);
453 
454  ast_debug(3, "App '%s' unsubscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
455  ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Unsubscribed\r\nAppName: %s\r\nToken: %s\r\n",
456  app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
457 }
#define AST_VECTOR_REMOVE_CMP_UNORDERED(vec, value, cmp, cleanup)
Remove an element from a vector that matches the given comparison.
Definition: vector.h:488
#define TECH_WILDCARD
Subscription to all technologies.
Definition: messaging.c:41
static ast_rwlock_t tech_subscriptions_lock
RWLock for tech_subscriptions.
Definition: messaging.c:78
static struct @498 tech_subscriptions
The subscriptions to technologies.
struct ast_endpoint * ast_endpoint_find_by_id(const char *id)
Finds the endpoint with the given tech[/resource] id.
const char * ast_endpoint_get_resource(const struct ast_endpoint *endpoint)
Gets the resource name of the given endpoint.
static struct ao2_container * endpoint_subscriptions
The subscriptions to endpoints.
Definition: messaging.c:67
#define AST_VECTOR_ELEM_CLEANUP_NOOP(elem)
Vector element cleanup that does nothing.
Definition: vector.h:571
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define ast_debug(level,...)
Log a DEBUG message.
#define ast_test_suite_event_notify(s, f,...)
Definition: test.h:189
#define ao2_unlink(container, obj)
Remove an object from a container.
Definition: astobj2.h:1578
const char * app_name(struct ast_app *app)
Definition: pbx_app.c:463
A subscription to some endpoint or technology.
Definition: messaging.c:59
#define RAII_VAR(vartype, varname, initval, dtor)
Declare a variable that will call a destructor function when it goes out of scope.
Definition: utils.h:941
#define AST_VECTOR_SIZE(vec)
Get the number of elements in a vector.
Definition: vector.h:609
const char * ast_endpoint_get_id(const struct ast_endpoint *endpoint)
Gets the tech/resource id of the given endpoint.
int messaging_cleanup ( void  )

Tidy up the messaging layer.

Return values
0success
-1failure

Definition at line 529 of file messaging.c.

References ao2_ref, ast_msg_handler_unregister(), AST_VECTOR_FREE, tech_subscriptions, and tech_subscriptions_lock.

530 {
531  ast_msg_handler_unregister(&ari_msg_handler);
534  ast_rwlock_destroy(&tech_subscriptions_lock);\
535 
536  return 0;
537 }
#define AST_VECTOR_FREE(vec)
Deallocates this vector.
Definition: vector.h:174
int ast_msg_handler_unregister(const struct ast_msg_handler *handler)
Unregister a ast_msg_handler.
static ast_rwlock_t tech_subscriptions_lock
RWLock for tech_subscriptions.
Definition: messaging.c:78
static struct @498 tech_subscriptions
The subscriptions to technologies.
static struct ao2_container * endpoint_subscriptions
The subscriptions to endpoints.
Definition: messaging.c:67
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
int messaging_init ( void  )

Initialize the messaging layer.

Return values
0success
-1failure

Definition at line 539 of file messaging.c.

References AO2_ALLOC_OPT_LOCK_RWLOCK, ao2_ref, ast_msg_handler_register(), ast_rwlock_init, AST_VECTOR_FREE, AST_VECTOR_INIT, ENDPOINTS_NUM_BUCKETS, message_subscription_compare_cb(), message_subscription_hash_cb(), tech_subscriptions, and tech_subscriptions_lock.

540 {
541  endpoint_subscriptions = ao2_t_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
543  message_subscription_compare_cb, "Endpoint messaging subscription container creation");
544  if (!endpoint_subscriptions) {
545  return -1;
546  }
547 
550  return -1;
551  }
552 
556  return -1;
557  }
558 
559  if (ast_msg_handler_register(&ari_msg_handler)) {
562  ast_rwlock_destroy(&tech_subscriptions_lock);
563  return -1;
564  }
565 
566  return 0;
567 }
#define AST_VECTOR_FREE(vec)
Deallocates this vector.
Definition: vector.h:174
static int message_subscription_compare_cb(void *obj, void *arg, int flags)
Definition: messaging.c:160
int ast_msg_handler_register(const struct ast_msg_handler *handler)
Register a ast_msg_handler.
#define ast_rwlock_init(rwlock)
wrapper for rwlock with tracking enabled
Definition: lock.h:224
static ast_rwlock_t tech_subscriptions_lock
RWLock for tech_subscriptions.
Definition: messaging.c:78
static struct @498 tech_subscriptions
The subscriptions to technologies.
#define AST_VECTOR_INIT(vec, size)
Initialize a vector.
Definition: vector.h:113
static struct ao2_container * endpoint_subscriptions
The subscriptions to endpoints.
Definition: messaging.c:67
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
static int message_subscription_hash_cb(const void *obj, const int flags)
Definition: messaging.c:138
#define ENDPOINTS_NUM_BUCKETS
Number of buckets for the endpoint_subscriptions container.
Definition: messaging.c:46

Variable Documentation

struct ast_msg_handler ari_msg_handler
Initial value:
= {
.name = "ari",
.handle_msg = handle_msg_cb,
.has_destination = has_destination_cb,
}

Definition at line 362 of file messaging.c.

struct { ... } tech_subscriptions

The subscriptions to technologies.

Note
These are stored separately from standard endpoints, given how relatively few of them there are.

Referenced by messaging_app_unsubscribe_endpoint(), messaging_cleanup(), and messaging_init().