Asterisk - The Open Source Telephony Project  21.4.1
res_stasis_device_state.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * Kevin Harwell <kharwell@digium.com>
7  *
8  * See http://www.asterisk.org for more information about
9  * the Asterisk project. Please do not directly contact
10  * any of the maintainers of this project for assistance;
11  * the project provides a web site, mailing lists and IRC
12  * channels for your use.
13  *
14  * This program is free software, distributed under the terms of
15  * the GNU General Public License Version 2. See the LICENSE file
16  * at the top of the source tree.
17  */
18 
19 /*** MODULEINFO
20  <depend type="module">res_stasis</depend>
21  <support_level>core</support_level>
22  ***/
23 
24 #include "asterisk.h"
25 
26 #include "asterisk/astdb.h"
27 #include "asterisk/astobj2.h"
28 #include "asterisk/module.h"
31 
32 #define DEVICE_STATE_SIZE 64
33 /*! astdb family name */
34 #define DEVICE_STATE_FAMILY "StasisDeviceState"
35 /*! Stasis device state provider */
36 #define DEVICE_STATE_PROVIDER_STASIS "Stasis"
37 /*! Scheme for custom device states */
38 #define DEVICE_STATE_SCHEME_STASIS "Stasis:"
39 /*! Scheme for device state subscriptions */
40 #define DEVICE_STATE_SCHEME_SUB "deviceState:"
41 
42 /*! Number of hash buckets for device state subscriptions */
43 #define DEVICE_STATE_BUCKETS 37
44 
45 /*! The key used for tracking a subscription to all device states */
46 #define DEVICE_STATE_ALL "__AST_DEVICE_STATE_ALL_TOPIC"
47 
48 /*! Container for subscribed device states */
49 static struct ao2_container *device_state_subscriptions;
50 
51 /*!
52  * \brief Device state subscription object.
53  */
56  AST_STRING_FIELD(app_name);
57  AST_STRING_FIELD(device_name);
58  );
59  /*! The subscription object */
61 };
62 
63 static int device_state_subscriptions_hash(const void *obj, const int flags)
64 {
65  const struct device_state_subscription *object;
66 
67  switch (flags & OBJ_SEARCH_MASK) {
68  case OBJ_SEARCH_OBJECT:
69  object = obj;
70  return ast_str_hash(object->device_name);
71  case OBJ_SEARCH_KEY:
72  default:
73  /* Hash can only work on something with a full key. */
74  ast_assert(0);
75  return 0;
76  }
77 }
78 
79 static int device_state_subscriptions_cmp(void *obj, void *arg, int flags)
80 {
81  const struct device_state_subscription *object_left = obj;
82  const struct device_state_subscription *object_right = arg;
83  int cmp;
84 
85  switch (flags & OBJ_SEARCH_MASK) {
86  case OBJ_SEARCH_OBJECT:
87  /* find objects matching both device and app names */
88  if (strcmp(object_left->device_name,
89  object_right->device_name)) {
90  return 0;
91  }
92  cmp = strcmp(object_left->app_name, object_right->app_name);
93  break;
94  case OBJ_SEARCH_KEY:
96  ast_assert(0); /* not supported by container */
97  /* fall through */
98  default:
99  cmp = 0;
100  break;
101  }
102 
103  return cmp ? 0 : CMP_MATCH | CMP_STOP;
104 }
105 
106 static void device_state_subscription_destroy(void *obj)
107 {
108  struct device_state_subscription *sub = obj;
110 }
111 
112 static struct device_state_subscription *device_state_subscription_create(
113  const struct stasis_app *app, const char *device_name)
114 {
116  const char *app_name = stasis_app_name(app);
117  size_t size;
118 
119  if (ast_strlen_zero(device_name)) {
120  device_name = DEVICE_STATE_ALL;
121  }
122 
123  size = strlen(device_name) + strlen(app_name) + 2;
124 
125  sub = ao2_alloc(sizeof(*sub), device_state_subscription_destroy);
126  if (!sub) {
127  return NULL;
128  }
129 
130  if (ast_string_field_init(sub, size)) {
131  ao2_ref(sub, -1);
132  return NULL;
133  }
134 
135  ast_string_field_set(sub, app_name, app_name);
136  ast_string_field_set(sub, device_name, device_name);
137  return sub;
138 }
139 
140 static struct device_state_subscription *find_device_state_subscription(
141  struct stasis_app *app, const char *name)
142 {
143  struct device_state_subscription dummy_sub = {
144  .app_name = stasis_app_name(app),
145  .device_name = name
146  };
147 
148  return ao2_find(device_state_subscriptions, &dummy_sub, OBJ_SEARCH_OBJECT | OBJ_NOLOCK);
149 }
150 
151 static void remove_device_state_subscription(
152  struct device_state_subscription *sub)
153 {
154  if (sub->sub) {
155  sub->sub = stasis_unsubscribe_and_join(sub->sub);
156  }
157  ao2_unlink_flags(device_state_subscriptions, sub, OBJ_NOLOCK);
158 }
159 
161  const char *name, enum ast_device_state state)
162 {
163  return ast_json_pack("{s: s, s: s}",
164  "name", name,
165  "state", ast_devstate_str(state));
166 }
167 
169 {
170  struct ast_json *array = ast_json_array_create();
171  struct ast_db_entry *tree;
172  struct ast_db_entry *entry;
173 
174  tree = ast_db_gettree(DEVICE_STATE_FAMILY, NULL);
175  for (entry = tree; entry; entry = entry->next) {
176  const char *name = strrchr(entry->key, '/');
177 
178  if (!ast_strlen_zero(name)) {
179  char device[DEVICE_STATE_SIZE];
180 
181  snprintf(device, sizeof(device), "%s%s", DEVICE_STATE_SCHEME_STASIS, ++name);
182  ast_json_array_append(array,
184  }
185  }
186  ast_db_freetree(tree);
187 
188  return array;
189 }
190 
191 static void send_device_state(struct device_state_subscription *sub,
192  const char *name, enum ast_device_state state)
193 {
194  RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
195 
196  json = ast_json_pack("{s:s, s:s, s:o, s:o}",
197  "type", "DeviceStateChanged",
198  "application", sub->app_name,
199  "timestamp", ast_json_timeval(ast_tvnow(), NULL),
200  "device_state", stasis_app_device_state_to_json(
201  name, state));
202 
203  if (!json) {
204  ast_log(LOG_ERROR, "Unable to create device state json object\n");
205  return;
206  }
207 
208  stasis_app_send(sub->app_name, json);
209 }
210 
212  const char *name, const char *value)
213 {
214  size_t size = strlen(DEVICE_STATE_SCHEME_STASIS);
215  enum ast_device_state state;
216 
217  ast_debug(3, "Updating device name = %s, value = %s", name, value);
218 
219  if (strncasecmp(name, DEVICE_STATE_SCHEME_STASIS, size)) {
220  ast_log(LOG_ERROR, "Update can only be used to set "
221  "'%s' device state!\n", DEVICE_STATE_SCHEME_STASIS);
223  }
224 
225  name += size;
226  if (ast_strlen_zero(name)) {
227  ast_log(LOG_ERROR, "Update requires custom device name!\n");
229  }
230 
231  if (!value || (state = ast_devstate_val(value)) == AST_DEVICE_UNKNOWN) {
232  ast_log(LOG_ERROR, "Unknown device state "
233  "value '%s'\n", value);
235  }
236 
237  ast_db_put(DEVICE_STATE_FAMILY, name, value);
239  DEVICE_STATE_SCHEME_STASIS, name);
240 
241  return STASIS_DEVICE_STATE_OK;
242 }
243 
245 {
246  const char *full_name = name;
247  size_t size = strlen(DEVICE_STATE_SCHEME_STASIS);
248 
249  if (strncasecmp(name, DEVICE_STATE_SCHEME_STASIS, size)) {
250  ast_log(LOG_ERROR, "Can only delete '%s' device states!\n",
251  DEVICE_STATE_SCHEME_STASIS);
253  }
254 
255  name += size;
256  if (ast_strlen_zero(name)) {
257  ast_log(LOG_ERROR, "Delete requires a device name!\n");
259  }
260 
261  if (ast_device_state_clear_cache(full_name)) {
263  }
264 
265  ast_db_del(DEVICE_STATE_FAMILY, name);
266 
267  /* send state change for delete */
270  DEVICE_STATE_SCHEME_STASIS, name);
271 
272  return STASIS_DEVICE_STATE_OK;
273 }
274 
275 static void populate_cache(void)
276 {
277  RAII_VAR(struct ast_db_entry *, tree,
278  ast_db_gettree(DEVICE_STATE_FAMILY, NULL), ast_db_freetree);
279  struct ast_db_entry *entry;
280 
281  for (entry = tree; entry; entry = entry->next) {
282  const char *name = strrchr(entry->key, '/');
283  if (!ast_strlen_zero(name)) {
285  ast_devstate_val(entry->data),
286  AST_DEVSTATE_CACHABLE, "%s%s\n",
287  DEVICE_STATE_SCHEME_STASIS, name + 1);
288  }
289  }
290 }
291 
292 static enum ast_device_state stasis_device_state_cb(const char *data)
293 {
294  char buf[DEVICE_STATE_SIZE];
295 
296  ast_db_get(DEVICE_STATE_FAMILY, data, buf, sizeof(buf));
297 
298  return ast_devstate_val(buf);
299 }
300 
301 static void device_state_cb(void *data, struct stasis_subscription *sub,
302  struct stasis_message *msg)
303 {
304  struct ast_device_state_message *device_state;
305 
306  if (stasis_subscription_final_message(sub, msg)) {
307  /* Remove stasis subscription's reference to device_state_subscription */
308  ao2_ref(data, -1);
309  return;
310  }
311 
313  return;
314  }
315 
316  device_state = stasis_message_data(msg);
317  if (device_state->eid) {
318  /* ignore non-aggregate states */
319  return;
320  }
321 
322  send_device_state(data, device_state->device, device_state->state);
323 }
324 
325 static void *find_device_state(const struct stasis_app *app, const char *name)
326 {
327  return device_state_subscription_create(app, name);
328 }
329 
330 static int is_subscribed_device_state(struct stasis_app *app, const char *name)
331 {
333 
334  sub = find_device_state_subscription(app, DEVICE_STATE_ALL);
335  if (sub) {
336  ao2_ref(sub, -1);
337  return 1;
338  }
339 
340  sub = find_device_state_subscription(app, name);
341  if (sub) {
342  ao2_ref(sub, -1);
343  return 1;
344  }
345 
346  return 0;
347 }
348 
349 static int is_subscribed_device_state_lock(struct stasis_app *app, const char *name)
350 {
351  int is_subscribed;
352 
353  ao2_lock(device_state_subscriptions);
354  is_subscribed = is_subscribed_device_state(app, name);
355  ao2_unlock(device_state_subscriptions);
356 
357  return is_subscribed;
358 }
359 
360 static int subscribe_device_state(struct stasis_app *app, void *obj)
361 {
362  struct device_state_subscription *sub = obj;
363  struct stasis_topic *topic;
364 
365  if (!sub) {
366  sub = device_state_subscription_create(app, NULL);
367  if (!sub) {
368  return -1;
369  }
370  }
371 
372  if (strcmp(sub->device_name, DEVICE_STATE_ALL)) {
373  topic = ast_device_state_topic(sub->device_name);
374  } else {
375  topic = ast_device_state_topic_all();
376  }
377 
378  ao2_lock(device_state_subscriptions);
379 
380  if (is_subscribed_device_state(app, sub->device_name)) {
381  ao2_unlock(device_state_subscriptions);
382  ast_debug(3, "App %s is already subscribed to %s\n", stasis_app_name(app), sub->device_name);
383  return 0;
384  }
385 
386  ast_debug(3, "Subscribing to device %s\n", sub->device_name);
387 
388  sub->sub = stasis_subscribe_pool(topic, device_state_cb, ao2_bump(sub));
389  if (!sub->sub) {
390  ao2_unlock(device_state_subscriptions);
391  ast_log(LOG_ERROR, "Unable to subscribe to device %s\n",
392  sub->device_name);
393  /* Reference we added when attempting to stasis_subscribe_pool */
394  ao2_ref(sub, -1);
395  return -1;
396  }
400 
401  ao2_link_flags(device_state_subscriptions, sub, OBJ_NOLOCK);
402  ao2_unlock(device_state_subscriptions);
403 
404  return 0;
405 }
406 
407 static int unsubscribe_device_state(struct stasis_app *app, const char *name)
408 {
410 
411  ao2_lock(device_state_subscriptions);
412  sub = find_device_state_subscription(app, name);
413  if (sub) {
414  remove_device_state_subscription(sub);
415  }
416  ao2_unlock(device_state_subscriptions);
417 
418  ao2_cleanup(sub);
419 
420  return 0;
421 }
422 
423 static int device_to_json_cb(void *obj, void *arg, void *data, int flags)
424 {
425  struct device_state_subscription *sub = obj;
426  const char *app_name = arg;
427  struct ast_json *array = data;
428 
429  if (strcmp(sub->app_name, app_name)) {
430  return 0;
431  }
432 
434  array, ast_json_string_create(sub->device_name));
435  return 0;
436 
437 }
438 
439 static void devices_to_json(const struct stasis_app *app, struct ast_json *json)
440 {
441  struct ast_json *array = ast_json_array_create();
442  ao2_callback_data(device_state_subscriptions, OBJ_NODATA,
443  device_to_json_cb, (void *)stasis_app_name(app), array);
444  ast_json_object_set(json, "device_names", array);
445 }
446 
447 struct stasis_app_event_source device_state_event_source = {
448  .scheme = DEVICE_STATE_SCHEME_SUB,
449  .find = find_device_state,
450  .subscribe = subscribe_device_state,
451  .unsubscribe = unsubscribe_device_state,
452  .is_subscribed = is_subscribed_device_state_lock,
453  .to_json = devices_to_json
454 };
455 
456 static int load_module(void)
457 {
458  populate_cache();
459  if (ast_devstate_prov_add(DEVICE_STATE_PROVIDER_STASIS,
460  stasis_device_state_cb)) {
462  }
463 
464  device_state_subscriptions = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
465  DEVICE_STATE_BUCKETS, device_state_subscriptions_hash, NULL,
466  device_state_subscriptions_cmp);
467  if (!device_state_subscriptions) {
468  ast_devstate_prov_del(DEVICE_STATE_PROVIDER_STASIS);
470  }
471 
472  stasis_app_register_event_source(&device_state_event_source);
474 }
475 
476 static int unload_module(void)
477 {
478  ast_devstate_prov_del(DEVICE_STATE_PROVIDER_STASIS);
479  stasis_app_unregister_event_source(&device_state_event_source);
480  ao2_cleanup(device_state_subscriptions);
481  device_state_subscriptions = NULL;
482  return 0;
483 }
484 
485 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Stasis application device state support",
486  .support_level = AST_MODULE_SUPPORT_CORE,
487  .load = load_module,
488  .unload = unload_module,
489  .requires = "res_stasis",
490 );
struct stasis_topic * ast_device_state_topic(const char *device)
Get the Stasis topic for device state messages for a specific device.
Definition: devicestate.c:683
Stasis Application Device State API. See StasisApplication API" for detailed documentation.
ast_device_state
Device States.
Definition: devicestate.h:52
Asterisk main include file. File version handling, generic pbx functions.
enum stasis_device_state_result stasis_app_device_state_delete(const char *name)
Delete a device controlled by ARI.
struct ast_json * ast_json_pack(char const *format,...)
Helper for creating complex JSON values.
Definition: json.c:612
The arg parameter is a search key, but is not an object.
Definition: astobj2.h:1101
void ast_json_unref(struct ast_json *value)
Decrease refcount on value. If refcount reaches zero, value is freed.
Definition: json.c:73
const char * scheme
The scheme to match against on [un]subscribes.
Definition: stasis_app.h:176
Device state subscription object.
void ast_db_freetree(struct ast_db_entry *entry)
Free structure created by ast_db_gettree()
Definition: main/db.c:677
stasis_device_state_result
enum ast_device_state state
Definition: devicestate.h:248
int ast_devstate_prov_del(const char *label)
Remove device state provider.
Definition: devicestate.c:418
Assume that the ao2_container is already locked.
Definition: astobj2.h:1063
struct stasis_message_type * stasis_message_type(const struct stasis_message *msg)
Get the message type for a stasis_message.
int stasis_subscription_set_filter(struct stasis_subscription *subscription, enum stasis_subscription_message_filter filter)
Set the message type filtering level on a subscription.
Definition: stasis.c:1077
int ast_devstate_prov_add(const char *label, ast_devstate_prov_cb_type callback)
Add device state provider.
Definition: devicestate.c:391
struct timeval ast_tvnow(void)
Returns current timeval. Meant to replace calls to gettimeofday().
Definition: time.h:159
#define AST_DECLARE_STRING_FIELDS(field_list)
Declare the fields needed in a structure.
Definition: stringfields.h:341
#define ao2_link_flags(container, obj, flags)
Add an object to a container.
Definition: astobj2.h:1554
int ast_json_object_set(struct ast_json *object, const char *key, struct ast_json *value)
Set a field in a JSON object.
Definition: json.c:414
int ast_device_state_clear_cache(const char *device)
Clear the device from the stasis cache.
Definition: devicestate.c:688
#define ao2_bump(obj)
Bump refcount on an AO2 object by one, returning the object.
Definition: astobj2.h:480
struct stasis_message_type * ast_device_state_message_type(void)
Get the Stasis message type for device state messages.
The arg parameter is a partial search key similar to OBJ_SEARCH_KEY.
Definition: astobj2.h:1116
int ast_devstate_changed(enum ast_device_state state, enum ast_devstate_cache cachable, const char *fmt,...)
Tells Asterisk the State for Device is changed.
Definition: devicestate.c:510
enum ast_device_state ast_devstate_val(const char *val)
Convert device state from text to integer value.
Definition: devicestate.c:260
struct ast_json * ast_json_string_create(const char *value)
Construct a JSON string from value.
Definition: json.c:278
const struct ast_eid * eid
The EID of the server where this message originated.
Definition: devicestate.h:246
#define ast_string_field_init(x, size)
Initialize a field pool and fields.
Definition: stringfields.h:359
Event source information and callbacks.
Definition: stasis_app.h:174
#define ao2_ref(o, delta)
Reference/unreference an object and return the old refcount.
Definition: astobj2.h:459
#define AST_STRING_FIELD(name)
Declare a string field.
Definition: stringfields.h:303
const char * stasis_app_name(const struct stasis_app *app)
Retrieve an application's name.
#define ast_debug(level,...)
Log a DEBUG message.
struct ast_json * ast_json_array_create(void)
Create a empty JSON array.
Definition: json.c:362
struct ast_json * ast_json_timeval(const struct timeval tv, const char *zone)
Construct a timeval as JSON.
Definition: json.c:670
int ast_json_array_append(struct ast_json *array, struct ast_json *value)
Append to an array.
Definition: json.c:378
void stasis_app_unregister_event_source(struct stasis_app_event_source *obj)
Unregister an application event source.
Definition: res_stasis.c:1830
const char * ast_devstate_str(enum ast_device_state devstate) attribute_pure
Convert device state to text string that is easier to parse.
Definition: devicestate.c:255
Backend API for implementing components of res_stasis.
#define ao2_container_alloc_hash(ao2_options, container_options, n_buckets, hash_fn, sort_fn, cmp_fn)
Allocate and initialize a hash container with the desired number of buckets.
Definition: astobj2.h:1303
struct ast_db_entry * ast_db_gettree(const char *family, const char *keytree)
Get a list of values within the astdb tree.
Definition: main/db.c:610
void * stasis_message_data(const struct stasis_message *msg)
Get the data contained in a message.
enum stasis_device_state_result stasis_app_device_state_update(const char *name, const char *value)
Changes the state of a device controlled by ARI.
const char * app_name(struct ast_app *app)
Definition: pbx_app.c:463
struct stasis_topic * ast_device_state_topic_all(void)
Get the Stasis topic for device state messages.
Definition: devicestate.c:668
Definition: astdb.h:31
int stasis_subscription_final_message(struct stasis_subscription *sub, struct stasis_message *msg)
Determine whether a message is the final message to be received on a subscription.
Definition: stasis.c:1174
static void device_state_cb(void *unused, struct stasis_subscription *sub, struct stasis_message *msg)
set a member's status based on device state of that member's interface
Definition: app_queue.c:2641
int stasis_app_send(const char *app_name, struct ast_json *message)
Send a message to the given Stasis application.
Definition: res_stasis.c:1663
struct stasis_subscription * stasis_unsubscribe_and_join(struct stasis_subscription *subscription)
Cancel a subscription, blocking until the last message is processed.
Definition: stasis.c:1134
Module has failed to load, may be in an inconsistent state.
Definition: module.h:78
int ast_db_get(const char *family, const char *key, char *value, int valuelen)
Get key value specified by family/key.
Definition: main/db.c:427
struct ast_json * stasis_app_device_states_to_json(void)
Convert device states to json array.
struct stasis_subscription * sub
The arg parameter is an object of the same type.
Definition: astobj2.h:1087
int ast_db_del(const char *family, const char *key)
Delete entry in astdb.
Definition: main/db.c:476
int stasis_subscription_accept_message_type(struct stasis_subscription *subscription, const struct stasis_message_type *type)
Indicate to a subscription that we are interested in a message type.
Definition: stasis.c:1023
#define ao2_unlink_flags(container, obj, flags)
Remove an object from a container.
Definition: astobj2.h:1600
Abstract JSON element (object, array, string, int, ...).
The structure that contains device state.
Definition: devicestate.h:238
Definition: search.h:40
int ast_db_put(const char *family, const char *key, const char *value)
Store value addressed by family/key.
Definition: main/db.c:342
struct stasis_message_type * stasis_subscription_change_type(void)
Gets the message type for subscription change notices.
Generic container type.
Search option field mask.
Definition: astobj2.h:1072
#define ASTERISK_GPL_KEY
The text the key() function should return.
Definition: module.h:46
struct ast_json * stasis_app_device_state_to_json(const char *name, enum ast_device_state state)
Convert device state to json.
Asterisk module definitions.
#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
Persistent data storage (akin to *doze registry)
#define ast_string_field_free_memory(x)
free all memory - to be called before destroying the object
Definition: stringfields.h:374
void stasis_app_register_event_source(struct stasis_app_event_source *obj)
Register an application event source.
Definition: res_stasis.c:1823
static force_inline int attribute_pure ast_str_hash(const char *str)
Compute a hash value on a string.
Definition: strings.h:1259
#define ast_string_field_set(x, field, data)
Set a field to a simple string value.
Definition: stringfields.h:521