Add multi-channel Stasis messages; refactor Dial AMI events to Stasis
[asterisk/asterisk.git] / main / stasis_channels.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * Matt Jordan <mjordan@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 Stasis Messages and Data Types for Channel Objects
22  *
23  * \author \verbatim Matt Jordan <mjordan@digium.com> \endverbatim
24  *
25  */
26
27 /*** MODULEINFO
28         <support_level>core</support_level>
29  ***/
30
31 #include "asterisk.h"
32
33 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
34
35 #include "asterisk/stasis.h"
36 #include "asterisk/astobj2.h"
37 #include "asterisk/stasis_channels.h"
38
39 #define NUM_MULTI_CHANNEL_BLOB_BUCKETS 7
40
41 /*! \brief Message type for channel snapshot messages */
42 static struct stasis_message_type *channel_snapshot_type;
43
44 /*! \brief Message type for channel blob messages */
45 static struct stasis_message_type *channel_blob_type;
46
47 /*! \brief Message type for channel dial messages */
48 static struct stasis_message_type *channel_dial_type;
49
50 /*! \brief Topic for all channels */
51 struct stasis_topic *channel_topic_all;
52
53 /*! \brief Caching topic for all channels */
54 struct stasis_caching_topic *channel_topic_all_cached;
55
56 struct stasis_message_type *ast_channel_dial_type(void)
57 {
58         return channel_dial_type;
59 }
60
61 struct stasis_message_type *ast_channel_blob_type(void)
62 {
63         return channel_blob_type;
64 }
65
66 struct stasis_message_type *ast_channel_snapshot_type(void)
67 {
68         return channel_snapshot_type;
69 }
70
71 struct stasis_topic *ast_channel_topic_all(void)
72 {
73         return channel_topic_all;
74 }
75
76 struct stasis_caching_topic *ast_channel_topic_all_cached(void)
77 {
78         return channel_topic_all_cached;
79 }
80
81 static const char *channel_snapshot_get_id(struct stasis_message *message)
82 {
83         struct ast_channel_snapshot *snapshot;
84         if (ast_channel_snapshot_type() != stasis_message_type(message)) {
85                 return NULL;
86         }
87         snapshot = stasis_message_data(message);
88         return snapshot->uniqueid;
89 }
90
91 /*! \internal \brief Hash function for \ref ast_channel_snapshot objects */
92 static int channel_snapshot_hash_cb(const void *obj, const int flags)
93 {
94         const struct ast_channel_snapshot *snapshot = obj;
95         const char *name = (flags & OBJ_KEY) ? obj : snapshot->name;
96         return ast_str_case_hash(name);
97 }
98
99 /*! \internal \brief Comparison function for \ref ast_channel_snapshot objects */
100 static int channel_snapshot_cmp_cb(void *obj, void *arg, int flags)
101 {
102         struct ast_channel_snapshot *left = obj;
103         struct ast_channel_snapshot *right = arg;
104         const char *match = (flags & OBJ_KEY) ? arg : right->name;
105         return strcasecmp(left->name, match) ? 0 : (CMP_MATCH | CMP_STOP);
106 }
107
108 static void channel_snapshot_dtor(void *obj)
109 {
110         struct ast_channel_snapshot *snapshot = obj;
111         ast_string_field_free_memory(snapshot);
112 }
113
114 struct ast_channel_snapshot *ast_channel_snapshot_create(struct ast_channel *chan)
115 {
116         RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
117
118         snapshot = ao2_alloc(sizeof(*snapshot), channel_snapshot_dtor);
119         if (!snapshot || ast_string_field_init(snapshot, 1024)) {
120                 return NULL;
121         }
122
123         ast_string_field_set(snapshot, name, ast_channel_name(chan));
124         ast_string_field_set(snapshot, accountcode, ast_channel_accountcode(chan));
125         ast_string_field_set(snapshot, peeraccount, ast_channel_peeraccount(chan));
126         ast_string_field_set(snapshot, userfield, ast_channel_userfield(chan));
127         ast_string_field_set(snapshot, uniqueid, ast_channel_uniqueid(chan));
128         ast_string_field_set(snapshot, linkedid, ast_channel_linkedid(chan));
129         ast_string_field_set(snapshot, parkinglot, ast_channel_parkinglot(chan));
130         ast_string_field_set(snapshot, hangupsource, ast_channel_hangupsource(chan));
131         if (ast_channel_appl(chan)) {
132                 ast_string_field_set(snapshot, appl, ast_channel_appl(chan));
133         }
134         if (ast_channel_data(chan)) {
135                 ast_string_field_set(snapshot, data, ast_channel_data(chan));
136         }
137         ast_string_field_set(snapshot, context, ast_channel_context(chan));
138         ast_string_field_set(snapshot, exten, ast_channel_exten(chan));
139
140         ast_string_field_set(snapshot, caller_name,
141                 S_COR(ast_channel_caller(chan)->id.name.valid, ast_channel_caller(chan)->id.name.str, ""));
142         ast_string_field_set(snapshot, caller_number,
143                 S_COR(ast_channel_caller(chan)->id.number.valid, ast_channel_caller(chan)->id.number.str, ""));
144
145         ast_string_field_set(snapshot, connected_name,
146                 S_COR(ast_channel_connected(chan)->id.name.valid, ast_channel_connected(chan)->id.name.str, ""));
147         ast_string_field_set(snapshot, connected_number,
148                 S_COR(ast_channel_connected(chan)->id.number.valid, ast_channel_connected(chan)->id.number.str, ""));
149
150         snapshot->creationtime = ast_channel_creationtime(chan);
151         snapshot->state = ast_channel_state(chan);
152         snapshot->priority = ast_channel_priority(chan);
153         snapshot->amaflags = ast_channel_amaflags(chan);
154         snapshot->hangupcause = ast_channel_hangupcause(chan);
155         snapshot->flags = *ast_channel_flags(chan);
156
157         ao2_ref(snapshot, +1);
158         return snapshot;
159 }
160
161 static void publish_message_for_channel_topics(struct stasis_message *message, struct ast_channel *chan)
162 {
163         if (chan) {
164                 stasis_publish(ast_channel_topic(chan), message);
165         } else {
166                 stasis_publish(ast_channel_topic_all(), message);
167         }
168 }
169
170 static void channel_blob_dtor(void *obj)
171 {
172         struct ast_channel_blob *event = obj;
173         ao2_cleanup(event->snapshot);
174         ast_json_unref(event->blob);
175 }
176
177 void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *peer, const char *dialstring, const char *dialstatus)
178 {
179         RAII_VAR(struct ast_multi_channel_blob *, payload, NULL, ao2_cleanup);
180         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
181         RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
182         struct ast_channel_snapshot *caller_snapshot;
183         struct ast_channel_snapshot *peer_snapshot;
184
185         ast_assert(peer != NULL);
186         blob = ast_json_pack("{s: s, s: s, s: s}",
187                              "type", "dial",
188                              "dialstatus", S_OR(dialstatus, ""),
189                              "dialstring", S_OR(dialstring, ""));
190         if (!blob) {
191                 return;
192         }
193         payload = ast_multi_channel_blob_create(blob);
194         if (!payload) {
195                 return;
196         }
197
198         if (caller) {
199                 caller_snapshot = ast_channel_snapshot_create(caller);
200                 if (!caller_snapshot) {
201                         return;
202                 }
203                 ast_multi_channel_blob_add_channel(payload, "caller", caller_snapshot);
204         }
205
206         peer_snapshot = ast_channel_snapshot_create(peer);
207         if (!peer_snapshot) {
208                 return;
209         }
210         ast_multi_channel_blob_add_channel(payload, "peer", peer_snapshot);
211
212         msg = stasis_message_create(ast_channel_dial_type(), payload);
213         if (!msg) {
214                 return;
215         }
216
217         publish_message_for_channel_topics(msg, caller);
218 }
219
220 struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
221                                                struct ast_json *blob)
222 {
223         RAII_VAR(struct ast_channel_blob *, obj, NULL, ao2_cleanup);
224         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
225         struct ast_json *type;
226
227         ast_assert(blob != NULL);
228
229         type = ast_json_object_get(blob, "type");
230         if (type == NULL) {
231                 ast_log(LOG_ERROR, "Invalid ast_channel_blob; missing type field\n");
232                 return NULL;
233         }
234
235         obj = ao2_alloc(sizeof(*obj), channel_blob_dtor);
236         if (!obj) {
237                 return NULL;
238         }
239
240         if (chan) {
241                 obj->snapshot = ast_channel_snapshot_create(chan);
242                 if (obj->snapshot == NULL) {
243                         return NULL;
244                 }
245         }
246
247         obj->blob = ast_json_ref(blob);
248
249         msg = stasis_message_create(ast_channel_blob_type(), obj);
250         if (!msg) {
251                 return NULL;
252         }
253
254         ao2_ref(msg, +1);
255         return msg;
256 }
257
258 const char *ast_channel_blob_json_type(struct ast_channel_blob *obj)
259 {
260         if (obj == NULL) {
261                 return NULL;
262         }
263
264         return ast_json_string_get(ast_json_object_get(obj->blob, "type"));
265 }
266
267 /*! \brief A channel snapshot wrapper object used in \ref ast_multi_channel_blob objects */
268 struct channel_role_snapshot {
269         struct ast_channel_snapshot *snapshot;  /*!< A channel snapshot */
270         char role[0];                                                   /*!< The role assigned to the channel */
271 };
272
273 /*! \brief A multi channel blob data structure for multi_channel_blob stasis messages */
274 struct ast_multi_channel_blob {
275         struct ao2_container *channel_snapshots;        /*!< A container holding the snapshots */
276         struct ast_json *blob;                                          /*< A blob of JSON data */
277 };
278
279 /*! \internal \brief Standard comparison function for \ref channel_role_snapshot objects */
280 static int channel_role_single_cmp_cb(void *obj, void *arg, int flags)
281 {
282         struct channel_role_snapshot *left = obj;
283         struct channel_role_snapshot *right = arg;
284         const char *match = (flags & OBJ_KEY) ? arg : right->role;
285         return strcasecmp(left->role, match) ? 0 : (CMP_MATCH | CMP_STOP);
286 }
287
288 /*! \internal \brief Multi comparison function for \ref channel_role_snapshot objects */
289 static int channel_role_multi_cmp_cb(void *obj, void *arg, int flags)
290 {
291         struct channel_role_snapshot *left = obj;
292         struct channel_role_snapshot *right = arg;
293         const char *match = (flags & OBJ_KEY) ? arg : right->role;
294         return strcasecmp(left->role, match) ? 0 : (CMP_MATCH);
295 }
296
297 /*! \internal \brief Hash function for \ref channel_role_snapshot objects */
298 static int channel_role_hash_cb(const void *obj, const int flags)
299 {
300         const struct channel_role_snapshot *snapshot = obj;
301         const char *name = (flags & OBJ_KEY) ? obj : snapshot->role;
302         return ast_str_case_hash(name);
303 }
304
305 /*! \internal \brief Destructor for \ref ast_multi_channel_blob objects */
306 static void multi_channel_blob_dtor(void *obj)
307 {
308         struct ast_multi_channel_blob *multi_blob = obj;
309
310         ao2_cleanup(multi_blob->channel_snapshots);
311         ast_json_unref(multi_blob->blob);
312 }
313
314 struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *blob)
315 {
316         RAII_VAR(struct ast_multi_channel_blob *, obj,
317                         ao2_alloc(sizeof(*obj), multi_channel_blob_dtor),
318                         ao2_cleanup);
319         struct ast_json *type;
320
321         ast_assert(blob != NULL);
322
323         if (!obj) {
324                 return NULL;
325         }
326
327         type = ast_json_object_get(blob, "type");
328         if (type == NULL) {
329                 ast_log(LOG_ERROR, "Invalid ast_multi_channel_blob; missing type field\n");
330                 return NULL;
331         }
332
333         obj->channel_snapshots = ao2_container_alloc(NUM_MULTI_CHANNEL_BLOB_BUCKETS,
334                         channel_role_hash_cb, channel_role_single_cmp_cb);
335         if (!obj->channel_snapshots) {
336                 return NULL;
337         }
338
339         obj->blob = ast_json_ref(blob);
340
341         ao2_ref(obj, +1);
342         return obj;
343 }
344
345 static void channel_role_snapshot_dtor(void *obj)
346 {
347         struct channel_role_snapshot *role_snapshot = obj;
348         ao2_cleanup(role_snapshot->snapshot);
349 }
350
351 void ast_multi_channel_blob_add_channel(struct ast_multi_channel_blob *obj, const char *role, struct ast_channel_snapshot *snapshot)
352 {
353         RAII_VAR(struct channel_role_snapshot *, role_snapshot, NULL, ao2_cleanup);
354         int role_len = strlen(role) + 1;
355
356         if (!obj || ast_strlen_zero(role) || !snapshot) {
357                 return;
358         }
359
360         role_snapshot = ao2_alloc(sizeof(*role_snapshot) + role_len, channel_role_snapshot_dtor);
361         if (!role_snapshot) {
362                 return;
363         }
364         ast_copy_string(role_snapshot->role, role, role_len);
365         role_snapshot->snapshot = snapshot;
366         ao2_ref(role_snapshot->snapshot, +1);
367         ao2_link(obj->channel_snapshots, role_snapshot);
368 }
369
370 struct ast_channel_snapshot *ast_multi_channel_blob_get_channel(struct ast_multi_channel_blob *obj, const char *role)
371 {
372         struct channel_role_snapshot *role_snapshot;
373
374         if (!obj || ast_strlen_zero(role)) {
375                 return NULL;
376         }
377         role_snapshot = ao2_find(obj->channel_snapshots, role, OBJ_KEY);
378         /* Note that this function does not increase the ref count on snapshot */
379         if (!role_snapshot) {
380                 return NULL;
381         }
382         ao2_ref(role_snapshot, -1);
383         return role_snapshot->snapshot;
384 }
385
386 struct ao2_container *ast_multi_channel_blob_get_channels(struct ast_multi_channel_blob *obj, const char *role)
387 {
388         RAII_VAR(struct ao2_container *, ret_container,
389                 ao2_container_alloc(NUM_MULTI_CHANNEL_BLOB_BUCKETS, channel_snapshot_hash_cb, channel_snapshot_cmp_cb),
390                 ao2_cleanup);
391         struct ao2_iterator *it_role_snapshots;
392         struct channel_role_snapshot *role_snapshot;
393         char *arg;
394
395         if (!obj || ast_strlen_zero(role) || !ret_container) {
396                 return NULL;
397         }
398         arg = ast_strdupa(role);
399
400         it_role_snapshots = ao2_callback(obj->channel_snapshots, OBJ_MULTIPLE | OBJ_KEY, channel_role_multi_cmp_cb, arg);
401         if (!it_role_snapshots) {
402                 return NULL;
403         }
404
405         while ((role_snapshot = ao2_iterator_next(it_role_snapshots))) {
406                 ao2_link(ret_container, role_snapshot->snapshot);
407                 ao2_ref(role_snapshot, -1);
408         }
409         ao2_iterator_destroy(it_role_snapshots);
410
411         ao2_ref(ret_container, +1);
412         return ret_container;
413 }
414
415 struct ast_json *ast_multi_channel_blob_get_json(struct ast_multi_channel_blob *obj)
416 {
417         if (!obj) {
418                 return NULL;
419         }
420         return obj->blob;
421 }
422
423 const char *ast_multi_channel_blob_get_type(struct ast_multi_channel_blob *obj)
424 {
425         if (!obj) {
426                 return NULL;
427         }
428
429         return ast_json_string_get(ast_json_object_get(obj->blob, "type"));
430 }
431
432 void ast_channel_publish_varset(struct ast_channel *chan, const char *name, const char *value)
433 {
434         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
435         RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
436
437         ast_assert(name != NULL);
438         ast_assert(value != NULL);
439
440         blob = ast_json_pack("{s: s, s: s, s: s}",
441                              "type", "varset",
442                              "variable", name,
443                              "value", value);
444         if (!blob) {
445                 ast_log(LOG_ERROR, "Error creating message\n");
446                 return;
447         }
448
449         msg = ast_channel_blob_create(chan, ast_json_ref(blob));
450
451         if (!msg) {
452                 return;
453         }
454
455         publish_message_for_channel_topics(msg, chan);
456 }
457
458 void ast_stasis_channels_shutdown(void)
459 {
460         ao2_cleanup(channel_snapshot_type);
461         channel_snapshot_type = NULL;
462         ao2_cleanup(channel_blob_type);
463         channel_blob_type = NULL;
464         ao2_cleanup(channel_dial_type);
465         channel_dial_type = NULL;
466         ao2_cleanup(channel_topic_all);
467         channel_topic_all = NULL;
468         channel_topic_all_cached = stasis_caching_unsubscribe(channel_topic_all_cached);
469 }
470
471 void ast_stasis_channels_init(void)
472 {
473         channel_snapshot_type = stasis_message_type_create("ast_channel_snapshot");
474         channel_blob_type = stasis_message_type_create("ast_channel_blob");
475         channel_dial_type = stasis_message_type_create("ast_channel_dial");
476         channel_topic_all = stasis_topic_create("ast_channel_topic_all");
477         channel_topic_all_cached = stasis_caching_topic_create(channel_topic_all, channel_snapshot_get_id);
478 }