res/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 ASTERISK_REGISTER_FILE()
  27
  28 #include "asterisk/astdb.h"
  29 #include "asterisk/astobj2.h"
  30 #include "asterisk/module.h"
  31 #include "asterisk/stasis_app_impl.h"
  32 #include "asterisk/stasis_app_device_state.h"
  33
  34 #define DEVICE_STATE_SIZE 64
  35 /*! astdb family name */
  36 #define DEVICE_STATE_FAMILY "StasisDeviceState"
  37 /*! Stasis device state provider */
  38 #define DEVICE_STATE_PROVIDER_STASIS "Stasis"
  39 /*! Scheme for custom device states */
  40 #define DEVICE_STATE_SCHEME_STASIS "Stasis:"
  41 /*! Scheme for device state subscriptions */
  42 #define DEVICE_STATE_SCHEME_SUB "deviceState:"
  43
  44 /*! Number of hash buckets for device state subscriptions */
  45 #define DEVICE_STATE_BUCKETS 37
  46
  47 /*! The key used for tracking a subscription to all device states */
  48 #define DEVICE_STATE_ALL "__AST_DEVICE_STATE_ALL_TOPIC"
  49
  50 /*! Container for subscribed device states */
  51 static struct ao2_container *device_state_subscriptions;
  52
  53 /*!
  54  * \brief Device state subscription object.
  55  */
  56 struct device_state_subscription {
  57         AST_DECLARE_STRING_FIELDS(
  58                 AST_STRING_FIELD(app_name);
  59                 AST_STRING_FIELD(device_name);
  60         );
  61         /*! The subscription object */
  62         struct stasis_subscription *sub;
  63 };
  64
  65 static int device_state_subscriptions_hash(const void *obj, const int flags)
  66 {
  67         const struct device_state_subscription *object;
  68
  69         switch (flags & OBJ_SEARCH_MASK) {
  70         case OBJ_SEARCH_OBJECT:
  71                 object = obj;
  72                 return ast_str_hash(object->device_name);
  73         case OBJ_SEARCH_KEY:
  74         default:
  75                 /* Hash can only work on something with a full key. */
  76                 ast_assert(0);
  77                 return 0;
  78         }
  79 }
  80
  81 static int device_state_subscriptions_cmp(void *obj, void *arg, int flags)
  82 {
  83         const struct device_state_subscription *object_left = obj;
  84         const struct device_state_subscription *object_right = arg;
  85         int cmp;
  86
  87         switch (flags & OBJ_SEARCH_MASK) {
  88         case OBJ_SEARCH_OBJECT:
  89                 /* find objects matching both device and app names */
  90                 if (strcmp(object_left->device_name,
  91                            object_right->device_name)) {
  92                         return 0;
  93                 }
  94                 cmp = strcmp(object_left->app_name, object_right->app_name);
  95                 break;
  96         case OBJ_SEARCH_KEY:
  97         case OBJ_SEARCH_PARTIAL_KEY:
  98                 ast_assert(0); /* not supported by container */
  99                 /* fall through */
 100         default:
 101                 cmp = 0;
 102                 break;
 103         }
 104
 105         return cmp ? 0 : CMP_MATCH | CMP_STOP;
 106 }
 107
 108 static void device_state_subscription_destroy(void *obj)
 109 {
 110         struct device_state_subscription *sub = obj;
 111         sub->sub = stasis_unsubscribe_and_join(sub->sub);
 112         ast_string_field_free_memory(sub);
 113 }
 114
 115 static struct device_state_subscription *device_state_subscription_create(
 116         const struct stasis_app *app, const char *device_name)
 117 {
 118         struct device_state_subscription *sub;
 119         const char *app_name = stasis_app_name(app);
 120         size_t size;
 121
 122         if (ast_strlen_zero(device_name)) {
 123                 device_name = DEVICE_STATE_ALL;
 124         }
 125
 126         size = strlen(device_name) + strlen(app_name) + 2;
 127
 128         sub = ao2_alloc(sizeof(*sub), device_state_subscription_destroy);
 129         if (!sub) {
 130                 return NULL;
 131         }
 132
 133         if (ast_string_field_init(sub, size)) {
 134                 ao2_ref(sub, -1);
 135                 return NULL;
 136         }
 137
 138         ast_string_field_set(sub, app_name, app_name);
 139         ast_string_field_set(sub, device_name, device_name);
 140         return sub;
 141 }
 142
 143 static struct device_state_subscription *find_device_state_subscription(
 144         struct stasis_app *app, const char *name)
 145 {
 146         struct device_state_subscription dummy_sub = {
 147                 .app_name = stasis_app_name(app),
 148                 .device_name = name
 149         };
 150
 151         return ao2_find(device_state_subscriptions, &dummy_sub, OBJ_SEARCH_OBJECT);
 152 }
 153
 154 static void remove_device_state_subscription(
 155         struct device_state_subscription *sub)
 156 {
 157         ao2_unlink(device_state_subscriptions, sub);
 158 }
 159
 160 struct ast_json *stasis_app_device_state_to_json(
 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
 168 struct ast_json *stasis_app_device_states_to_json(void)
 169 {
 170         struct ast_json *array = ast_json_array_create();
 171         RAII_VAR(struct ast_db_entry *, tree,
 172                  ast_db_gettree(DEVICE_STATE_FAMILY, NULL), ast_db_freetree);
 173         struct ast_db_entry *entry;
 174
 175         for (entry = tree; entry; entry = entry->next) {
 176                 const char *name = strrchr(entry->key, '/');
 177                 if (!ast_strlen_zero(name)) {
 178                         struct ast_str *device = ast_str_alloca(DEVICE_STATE_SIZE);
 179                         ast_str_set(&device, 0, "%s%s",
 180                                     DEVICE_STATE_SCHEME_STASIS, ++name);
 181                         ast_json_array_append(
 182                                 array, stasis_app_device_state_to_json(
 183                                         ast_str_buffer(device),
 184                                         ast_device_state(ast_str_buffer(device))));
 185                 }
 186         }
 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
 211 enum stasis_device_state_result stasis_app_device_state_update(
 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);
 222                 return STASIS_DEVICE_STATE_NOT_CONTROLLED;
 223         }
 224
 225         name += size;
 226         if (ast_strlen_zero(name)) {
 227                 ast_log(LOG_ERROR, "Update requires custom device name!\n");
 228                 return STASIS_DEVICE_STATE_MISSING;
 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);
 234                 return STASIS_DEVICE_STATE_UNKNOWN;
 235         }
 236
 237         ast_db_put(DEVICE_STATE_FAMILY, name, value);
 238         ast_devstate_changed(state, AST_DEVSTATE_CACHABLE, "%s%s",
 239                              DEVICE_STATE_SCHEME_STASIS, name);
 240
 241         return STASIS_DEVICE_STATE_OK;
 242 }
 243
 244 enum stasis_device_state_result stasis_app_device_state_delete(const char *name)
 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);
 252                 return STASIS_DEVICE_STATE_NOT_CONTROLLED;
 253         }
 254
 255         name += size;
 256         if (ast_strlen_zero(name)) {
 257                 ast_log(LOG_ERROR, "Delete requires a device name!\n");
 258                 return STASIS_DEVICE_STATE_MISSING;
 259         }
 260
 261         if (ast_device_state_clear_cache(full_name)) {
 262                 return STASIS_DEVICE_STATE_UNKNOWN;
 263         }
 264
 265         ast_db_del(DEVICE_STATE_FAMILY, name);
 266
 267         /* send state change for delete */
 268         ast_devstate_changed(
 269                 AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "%s%s",
 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)) {
 284                         ast_devstate_changed(
 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
 312         if (ast_device_state_message_type() != stasis_message_type(msg)) {
 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 {
 332         struct device_state_subscription *sub;
 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 subscribe_device_state(struct stasis_app *app, void *obj)
 350 {
 351         struct device_state_subscription *sub = obj;
 352         struct stasis_topic *topic;
 353
 354         if (!sub) {
 355                 sub = device_state_subscription_create(app, NULL);
 356                 if (!sub) {
 357                         return -1;
 358                 }
 359         }
 360
 361         if (strcmp(sub->device_name, DEVICE_STATE_ALL)) {
 362                 topic = ast_device_state_topic(sub->device_name);
 363         } else {
 364                 topic = ast_device_state_topic_all();
 365         }
 366
 367         if (is_subscribed_device_state(app, sub->device_name)) {
 368                 ast_debug(3, "App %s is already subscribed to %s\n", stasis_app_name(app), sub->device_name);
 369                 return 0;
 370         }
 371
 372         ast_debug(3, "Subscribing to device %s\n", sub->device_name);
 373
 374         sub->sub = stasis_subscribe_pool(topic, device_state_cb, ao2_bump(sub));
 375         if (!sub->sub) {
 376                 ast_log(LOG_ERROR, "Unable to subscribe to device %s\n",
 377                         sub->device_name);
 378                 /* Reference we added when attempting to stasis_subscribe_pool */
 379                 ao2_ref(sub, -1);
 380                 return -1;
 381         }
 382
 383         ao2_link(device_state_subscriptions, sub);
 384         return 0;
 385 }
 386
 387 static int unsubscribe_device_state(struct stasis_app *app, const char *name)
 388 {
 389         RAII_VAR(struct device_state_subscription *, sub,
 390                  find_device_state_subscription(app, name), ao2_cleanup);
 391         remove_device_state_subscription(sub);
 392         return 0;
 393 }
 394
 395 static int device_to_json_cb(void *obj, void *arg, void *data, int flags)
 396 {
 397         struct device_state_subscription *sub = obj;
 398         const char *app_name = arg;
 399         struct ast_json *array = data;
 400
 401         if (strcmp(sub->app_name, app_name)) {
 402                 return 0;
 403         }
 404
 405         ast_json_array_append(
 406                 array, ast_json_string_create(sub->device_name));
 407         return 0;
 408
 409 }
 410
 411 static void devices_to_json(const struct stasis_app *app, struct ast_json *json)
 412 {
 413         struct ast_json *array = ast_json_array_create();
 414         ao2_callback_data(device_state_subscriptions, OBJ_NODATA,
 415                           device_to_json_cb, (void *)stasis_app_name(app), array);
 416         ast_json_object_set(json, "device_names", array);
 417 }
 418
 419 struct stasis_app_event_source device_state_event_source = {
 420         .scheme = DEVICE_STATE_SCHEME_SUB,
 421         .find = find_device_state,
 422         .subscribe = subscribe_device_state,
 423         .unsubscribe = unsubscribe_device_state,
 424         .is_subscribed = is_subscribed_device_state,
 425         .to_json = devices_to_json
 426 };
 427
 428 static int load_module(void)
 429 {
 430         populate_cache();
 431         if (ast_devstate_prov_add(DEVICE_STATE_PROVIDER_STASIS,
 432                                   stasis_device_state_cb)) {
 433                 return AST_MODULE_LOAD_FAILURE;
 434         }
 435
 436         if (!(device_state_subscriptions = ao2_container_alloc(
 437                       DEVICE_STATE_BUCKETS, device_state_subscriptions_hash,
 438                       device_state_subscriptions_cmp))) {
 439                 return AST_MODULE_LOAD_FAILURE;
 440         }
 441
 442         stasis_app_register_event_source(&device_state_event_source);
 443         return AST_MODULE_LOAD_SUCCESS;
 444 }
 445
 446 static int unload_module(void)
 447 {
 448         ast_devstate_prov_del(DEVICE_STATE_PROVIDER_STASIS);
 449         stasis_app_unregister_event_source(&device_state_event_source);
 450         ao2_cleanup(device_state_subscriptions);
 451         device_state_subscriptions = NULL;
 452         return 0;
 453 }
 454
 455 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS, "Stasis application device state support",
 456         .support_level = AST_MODULE_SUPPORT_CORE,
 457         .load = load_module,
 458         .unload = unload_module,
 459         .nonoptreq = "res_stasis"
 460 );