main/endpoints.c

   1 /*
   2  * Asterisk -- An open source telephony toolkit.
   3  *
   4  * Copyright (C) 2013, Digium, Inc.
   5  *
   6  * David M. Lee, II <dlee@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 /*! \file
  20  *
  21  * \brief Asterisk endpoint API.
  22  *
  23  * \author David M. Lee, II <dlee@digium.com>
  24  */
  25
  26 /*** MODULEINFO
  27         <support_level>core</support_level>
  28  ***/
  29
  30 #include "asterisk.h"
  31
  32 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
  33
  34 #include "asterisk/astobj2.h"
  35 #include "asterisk/endpoints.h"
  36 #include "asterisk/stasis.h"
  37 #include "asterisk/stasis_channels.h"
  38 #include "asterisk/stasis_endpoints.h"
  39 #include "asterisk/stasis_message_router.h"
  40 #include "asterisk/stringfields.h"
  41
  42 /*! Buckets for endpoint->channel mappings. Keep it prime! */
  43 #define ENDPOINT_BUCKETS 127
  44
  45 struct ast_endpoint {
  46         AST_DECLARE_STRING_FIELDS(
  47                 AST_STRING_FIELD(tech); /*!< Technology (SIP, IAX2, etc.). */
  48                 AST_STRING_FIELD(resource);     /*!< Name, unique to the tech. */
  49                 AST_STRING_FIELD(id);   /*!< tech/resource id */
  50                 );
  51         /*! Endpoint's current state */
  52         enum ast_endpoint_state state;
  53         /*!
  54          * \brief Max channels for this endpoint. -1 means unlimited or unknown.
  55          *
  56          * Note that this simply documents the limits of an endpoint, and does
  57          * nothing to try to enforce the limit.
  58          */
  59         int max_channels;
  60         /*! Topic for this endpoint's messages */
  61         struct stasis_topic *topic;
  62         /*!
  63          * Forwarding subscription sending messages to ast_endpoint_topic_all()
  64          */
  65         struct stasis_subscription *forward;
  66         /*! Router for handling this endpoint's messages */
  67         struct stasis_message_router *router;
  68         /*! ast_str_container of channels associated with this endpoint */
  69         struct ao2_container *channel_ids;
  70 };
  71
  72 const char *ast_endpoint_state_to_string(enum ast_endpoint_state state)
  73 {
  74         switch (state) {
  75         case AST_ENDPOINT_UNKNOWN:
  76                 return "unknown";
  77         case AST_ENDPOINT_OFFLINE:
  78                 return "offline";
  79         case AST_ENDPOINT_ONLINE:
  80                 return "online";
  81         }
  82         return "?";
  83 }
  84
  85 static void endpoint_publish_snapshot(struct ast_endpoint *endpoint)
  86 {
  87         RAII_VAR(struct ast_endpoint_snapshot *, snapshot, NULL, ao2_cleanup);
  88         RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
  89
  90         ast_assert(endpoint != NULL);
  91         ast_assert(endpoint->topic != NULL);
  92
  93         snapshot = ast_endpoint_snapshot_create(endpoint);
  94         if (!snapshot) {
  95                 return;
  96         }
  97         message = stasis_message_create(ast_endpoint_snapshot_type(), snapshot);
  98         if (!message) {
  99                 return;
 100         }
 101         stasis_publish(endpoint->topic, message);
 102 }
 103
 104 static void endpoint_dtor(void *obj)
 105 {
 106         struct ast_endpoint *endpoint = obj;
 107
 108         /* The router should be shut down already */
 109         ast_assert(stasis_message_router_is_done(endpoint->router));
 110         ao2_cleanup(endpoint->router);
 111         endpoint->router = NULL;
 112
 113         stasis_unsubscribe(endpoint->forward);
 114         endpoint->forward = NULL;
 115
 116         ao2_cleanup(endpoint->topic);
 117         endpoint->topic = NULL;
 118
 119         ast_string_field_free_memory(endpoint);
 120 }
 121
 122 static void endpoint_channel_snapshot(void *data,
 123         struct stasis_subscription *sub, struct stasis_topic *topic,
 124         struct stasis_message *message)
 125 {
 126         struct ast_endpoint *endpoint = data;
 127         struct ast_channel_snapshot *snapshot = stasis_message_data(message);
 128         RAII_VAR(char *, existing_id, NULL, ao2_cleanup);
 129         int publish = 0;
 130
 131         ast_assert(endpoint != NULL);
 132         ast_assert(snapshot != NULL);
 133
 134         ao2_lock(endpoint);
 135         existing_id = ao2_find(endpoint->channel_ids, snapshot->uniqueid,
 136                 OBJ_POINTER);
 137         if (!existing_id) {
 138                 ast_str_container_add(endpoint->channel_ids,
 139                         snapshot->uniqueid);
 140                 publish = 1;
 141         }
 142         ao2_unlock(endpoint);
 143         if (publish) {
 144                 endpoint_publish_snapshot(endpoint);
 145         }
 146 }
 147
 148 /*! \brief Handler for channel snapshot cache clears */
 149 static void endpoint_cache_clear(void *data,
 150         struct stasis_subscription *sub, struct stasis_topic *topic,
 151         struct stasis_message *message)
 152 {
 153         struct ast_endpoint *endpoint = data;
 154         struct stasis_message *clear_msg = stasis_message_data(message);
 155         struct ast_channel_snapshot *clear_snapshot;
 156
 157         if (stasis_message_type(clear_msg) != ast_channel_snapshot_type()) {
 158                 return;
 159         }
 160
 161         clear_snapshot = stasis_message_data(clear_msg);
 162
 163         ast_assert(endpoint != NULL);
 164
 165         ao2_lock(endpoint);
 166         ast_str_container_remove(endpoint->channel_ids, clear_snapshot->uniqueid);
 167         ao2_unlock(endpoint);
 168         endpoint_publish_snapshot(endpoint);
 169 }
 170
 171 static void endpoint_default(void *data,
 172         struct stasis_subscription *sub, struct stasis_topic *topic,
 173         struct stasis_message *message)
 174 {
 175         struct stasis_endpoint *endpoint = data;
 176
 177         if (stasis_subscription_final_message(sub, message)) {
 178                 ao2_cleanup(endpoint);
 179         }
 180 }
 181
 182 struct ast_endpoint *ast_endpoint_create(const char *tech, const char *resource)
 183 {
 184         RAII_VAR(struct ast_endpoint *, endpoint, NULL, ao2_cleanup);
 185         int r = 0;
 186
 187         if (ast_strlen_zero(tech)) {
 188                 ast_log(LOG_ERROR, "Endpoint tech cannot be empty\n");
 189                 return NULL;
 190         }
 191
 192         if (ast_strlen_zero(resource)) {
 193                 ast_log(LOG_ERROR, "Endpoint resource cannot be empty\n");
 194                 return NULL;
 195         }
 196
 197         endpoint = ao2_alloc(sizeof(*endpoint), endpoint_dtor);
 198         if (!endpoint) {
 199                 return NULL;
 200         }
 201
 202         endpoint->max_channels = -1;
 203         endpoint->state = AST_ENDPOINT_UNKNOWN;
 204
 205         if (ast_string_field_init(endpoint, 80) != 0) {
 206                 return NULL;
 207         }
 208
 209         ast_string_field_set(endpoint, tech, tech);
 210         ast_string_field_set(endpoint, resource, resource);
 211         ast_string_field_build(endpoint, id, "%s/%s", tech, resource);
 212
 213         /* All access to channel_ids should be covered by the endpoint's
 214          * lock; no extra lock needed. */
 215         endpoint->channel_ids = ast_str_container_alloc_options(
 216                 AO2_ALLOC_OPT_LOCK_NOLOCK, ENDPOINT_BUCKETS);
 217         if (!endpoint->channel_ids) {
 218                 return NULL;
 219         }
 220
 221         endpoint->topic = stasis_topic_create(endpoint->id);
 222         if (!endpoint->topic) {
 223                 return NULL;
 224         }
 225
 226         endpoint->forward =
 227                 stasis_forward_all(endpoint->topic, ast_endpoint_topic_all());
 228         if (!endpoint->forward) {
 229                 return NULL;
 230         }
 231
 232         endpoint->router = stasis_message_router_create(endpoint->topic);
 233         if (!endpoint->router) {
 234                 return NULL;
 235         }
 236         r |= stasis_message_router_add(endpoint->router,
 237                 ast_channel_snapshot_type(), endpoint_channel_snapshot,
 238                 endpoint);
 239         r |= stasis_message_router_add(endpoint->router,
 240                 stasis_cache_clear_type(), endpoint_cache_clear,
 241                 endpoint);
 242         r |= stasis_message_router_set_default(endpoint->router,
 243                 endpoint_default, endpoint);
 244
 245         endpoint_publish_snapshot(endpoint);
 246
 247         ao2_ref(endpoint, +1);
 248         return endpoint;
 249 }
 250
 251 const char *ast_endpoint_get_tech(const struct ast_endpoint *endpoint)
 252 {
 253         ast_assert(endpoint != NULL);
 254         return endpoint->tech;
 255 }
 256
 257 static struct stasis_message *create_endpoint_snapshot_message(struct ast_endpoint *endpoint)
 258 {
 259         RAII_VAR(struct ast_endpoint_snapshot *, snapshot, NULL, ao2_cleanup);
 260         snapshot = ast_endpoint_snapshot_create(endpoint);
 261         if (!snapshot) {
 262                 return NULL;
 263         }
 264
 265         return stasis_message_create(ast_endpoint_snapshot_type(), snapshot);
 266 }
 267
 268 void ast_endpoint_shutdown(struct ast_endpoint *endpoint)
 269 {
 270         RAII_VAR(struct stasis_message *, clear_msg, NULL, ao2_cleanup);
 271
 272         if (endpoint == NULL) {
 273                 return;
 274         }
 275
 276         clear_msg = create_endpoint_snapshot_message(endpoint);
 277         if (clear_msg) {
 278                 RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
 279                 message = stasis_cache_clear_create(clear_msg);
 280                 if (message) {
 281                         stasis_publish(endpoint->topic, message);
 282                 }
 283         }
 284
 285         /* Bump refcount to hold on to the router */
 286         ao2_ref(endpoint->router, +1);
 287         stasis_message_router_unsubscribe(endpoint->router);
 288 }
 289
 290 const char *ast_endpoint_get_resource(const struct ast_endpoint *endpoint)
 291 {
 292         return endpoint->resource;
 293 }
 294
 295 struct stasis_topic *ast_endpoint_topic(struct ast_endpoint *endpoint)
 296 {
 297         return endpoint ? endpoint->topic : ast_endpoint_topic_all();
 298 }
 299
 300 void ast_endpoint_set_state(struct ast_endpoint *endpoint,
 301         enum ast_endpoint_state state)
 302 {
 303         ast_assert(endpoint != NULL);
 304         ao2_lock(endpoint);
 305         endpoint->state = state;
 306         ao2_unlock(endpoint);
 307         endpoint_publish_snapshot(endpoint);
 308 }
 309
 310 void ast_endpoint_set_max_channels(struct ast_endpoint *endpoint,
 311         int max_channels)
 312 {
 313         ast_assert(endpoint != NULL);
 314         ao2_lock(endpoint);
 315         endpoint->max_channels = max_channels;
 316         ao2_unlock(endpoint);
 317         endpoint_publish_snapshot(endpoint);
 318 }
 319
 320 static void endpoint_snapshot_dtor(void *obj)
 321 {
 322         struct ast_endpoint_snapshot *snapshot = obj;
 323
 324         ast_assert(snapshot != NULL);
 325         ast_string_field_free_memory(snapshot);
 326 }
 327
 328 struct ast_endpoint_snapshot *ast_endpoint_snapshot_create(
 329         struct ast_endpoint *endpoint)
 330 {
 331         RAII_VAR(struct ast_endpoint_snapshot *, snapshot, NULL, ao2_cleanup);
 332         int channel_count;
 333         struct ao2_iterator i;
 334         void *obj;
 335         SCOPED_AO2LOCK(lock, endpoint);
 336
 337         channel_count = ao2_container_count(endpoint->channel_ids);
 338
 339         snapshot = ao2_alloc(
 340                 sizeof(*snapshot) + channel_count * sizeof(char *),
 341                 endpoint_snapshot_dtor);
 342
 343         if (ast_string_field_init(snapshot, 80) != 0) {
 344                 return NULL;
 345         }
 346
 347         ast_string_field_build(snapshot, id, "%s/%s", endpoint->tech,
 348                 endpoint->resource);
 349         ast_string_field_set(snapshot, tech, endpoint->tech);
 350         ast_string_field_set(snapshot, resource, endpoint->resource);
 351
 352         snapshot->state = endpoint->state;
 353         snapshot->max_channels = endpoint->max_channels;
 354
 355         i = ao2_iterator_init(endpoint->channel_ids, 0);
 356         while ((obj = ao2_iterator_next(&i))) {
 357                 RAII_VAR(char *, channel_id, obj, ao2_cleanup);
 358                 snapshot->channel_ids[snapshot->num_channels++] = channel_id;
 359         }
 360
 361         ao2_ref(snapshot, +1);
 362         return snapshot;
 363 }