Split caching out from the stasis_caching_topic.
[asterisk/asterisk.git] / res / stasis / control.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 Stasis application control support.
22  *
23  * \author David M. Lee, II <dlee@digium.com>
24  */
25
26 #include "asterisk.h"
27
28 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
29
30 #include "asterisk/stasis_channels.h"
31
32 #include "command.h"
33 #include "control.h"
34 #include "asterisk/dial.h"
35 #include "asterisk/bridge.h"
36 #include "asterisk/bridge_basic.h"
37 #include "asterisk/bridge_features.h"
38 #include "asterisk/frame.h"
39 #include "asterisk/pbx.h"
40 #include "asterisk/musiconhold.h"
41
42 struct stasis_app_control {
43         /*! Queue of commands to dispatch on the channel */
44         struct ao2_container *command_queue;
45         /*!
46          * When set, /c app_stasis should exit and continue in the dialplan.
47          */
48         int is_done:1;
49         /*!
50          * The associated channel.
51          * Be very careful with the threading associated w/ manipulating
52          * the channel.
53          */
54         struct ast_channel *channel;
55 };
56
57 struct stasis_app_control *control_create(struct ast_channel *channel)
58 {
59         struct stasis_app_control *control;
60
61         control = ao2_alloc(sizeof(*control), NULL);
62         if (!control) {
63                 return NULL;
64         }
65
66         control->command_queue = ao2_container_alloc_list(
67                 AO2_ALLOC_OPT_LOCK_MUTEX, 0, NULL, NULL);
68
69         if (!control->command_queue) {
70                 ao2_cleanup(control);
71                 return NULL;
72         }
73
74         control->channel = channel;
75
76         return control;
77 }
78
79 static struct stasis_app_command *exec_command(
80         struct stasis_app_control *control, stasis_app_command_cb command_fn,
81         void *data)
82 {
83         RAII_VAR(struct stasis_app_command *, command, NULL, ao2_cleanup);
84
85         command = command_create(command_fn, data);
86
87         if (!command) {
88                 return NULL;
89         }
90
91         /* command_queue is a thread safe list; no lock needed */
92         ao2_link(control->command_queue, command);
93
94         ao2_ref(command, +1);
95         return command;
96 }
97
98 struct stasis_app_control_dial_data {
99         char endpoint[AST_CHANNEL_NAME];
100         int timeout;
101 };
102
103 static void *app_control_dial(struct stasis_app_control *control,
104         struct ast_channel *chan, void *data)
105 {
106         RAII_VAR(struct ast_dial *, dial, ast_dial_create(), ast_dial_destroy);
107         RAII_VAR(struct stasis_app_control_dial_data *, dial_data, data, ast_free);
108         enum ast_dial_result res;
109         char *tech, *resource;
110
111         struct ast_channel *new_chan;
112         struct ast_bridge *bridge;
113
114         tech = dial_data->endpoint;
115         if (!(resource = strchr(tech, '/'))) {
116                 return NULL;
117         }
118         *resource++ = '\0';
119
120         if (!dial) {
121                 ast_log(LOG_ERROR, "Failed to create dialing structure.\n");
122                 return NULL;
123         }
124
125         if (ast_dial_append(dial, tech, resource) < 0) {
126                 ast_log(LOG_ERROR, "Failed to add %s/%s to dialing structure.\n", tech, resource);
127                 return NULL;
128         }
129
130         ast_dial_set_global_timeout(dial, dial_data->timeout);
131
132         res = ast_dial_run(dial, NULL, 0);
133
134         if (res != AST_DIAL_RESULT_ANSWERED || !(new_chan = ast_dial_answered_steal(dial))) {
135                 return NULL;
136         }
137
138         if (!(bridge = ast_bridge_basic_new())) {
139                 ast_log(LOG_ERROR, "Failed to create basic bridge.\n");
140                 return NULL;
141         }
142
143         ast_bridge_impart(bridge, new_chan, NULL, NULL, 1);
144         stasis_app_control_add_channel_to_bridge(control, bridge);
145
146         return NULL;
147 }
148
149 int stasis_app_control_dial(struct stasis_app_control *control, const char *endpoint, int timeout)
150 {
151         struct stasis_app_control_dial_data *dial_data;
152
153         if (!(dial_data = ast_calloc(1, sizeof(*dial_data)))) {
154                 return -1;
155         }
156
157         ast_copy_string(dial_data->endpoint, endpoint, sizeof(dial_data->endpoint));
158
159         if (timeout > 0) {
160                 dial_data->timeout = timeout * 1000;
161         } else if (timeout == -1) {
162                 dial_data->timeout = -1;
163         } else {
164                 dial_data->timeout = 30000;
165         }
166
167         stasis_app_send_command_async(control, app_control_dial, dial_data);
168
169         return 0;
170 }
171
172 int control_is_done(struct stasis_app_control *control)
173 {
174         /* Called from stasis_app_exec thread; no lock needed */
175         return control->is_done;
176 }
177
178 struct stasis_app_control_continue_data {
179         char context[AST_MAX_CONTEXT];
180         char extension[AST_MAX_EXTENSION];
181         int priority;
182 };
183
184 static void *app_control_continue(struct stasis_app_control *control,
185         struct ast_channel *chan, void *data)
186 {
187         RAII_VAR(struct stasis_app_control_continue_data *, continue_data, data, ast_free);
188
189         /* Called from stasis_app_exec thread; no lock needed */
190         ast_explicit_goto(control->channel, continue_data->context, continue_data->extension, continue_data->priority);
191
192         control->is_done = 1;
193
194         return NULL;
195 }
196
197 int stasis_app_control_continue(struct stasis_app_control *control, const char *context, const char *extension, int priority)
198 {
199         struct stasis_app_control_continue_data *continue_data;
200
201         if (!(continue_data = ast_calloc(1, sizeof(*continue_data)))) {
202                 return -1;
203         }
204         ast_copy_string(continue_data->context, S_OR(context, ""), sizeof(continue_data->context));
205         ast_copy_string(continue_data->extension, S_OR(extension, ""), sizeof(continue_data->extension));
206         if (priority > 0) {
207                 continue_data->priority = priority;
208         } else {
209                 continue_data->priority = -1;
210         }
211
212         stasis_app_send_command_async(control, app_control_continue, continue_data);
213
214         return 0;
215 }
216
217 struct stasis_app_control_mute_data {
218         enum ast_frame_type frametype;
219         unsigned int direction;
220 };
221
222 static void *app_control_mute(struct stasis_app_control *control,
223         struct ast_channel *chan, void *data)
224 {
225         RAII_VAR(struct stasis_app_control_mute_data *, mute_data, data, ast_free);
226         SCOPED_CHANNELLOCK(lockvar, chan);
227
228         ast_channel_suppress(control->channel, mute_data->direction, mute_data->frametype);
229
230         return NULL;
231 }
232
233 int stasis_app_control_mute(struct stasis_app_control *control, unsigned int direction, enum ast_frame_type frametype)
234 {
235         struct stasis_app_control_mute_data *mute_data;
236
237         if (!(mute_data = ast_calloc(1, sizeof(*mute_data)))) {
238                 return -1;
239         }
240
241         mute_data->direction = direction;
242         mute_data->frametype = frametype;
243
244         stasis_app_send_command_async(control, app_control_mute, mute_data);
245
246         return 0;
247 }
248
249 static void *app_control_unmute(struct stasis_app_control *control,
250         struct ast_channel *chan, void *data)
251 {
252         RAII_VAR(struct stasis_app_control_mute_data *, mute_data, data, ast_free);
253         SCOPED_CHANNELLOCK(lockvar, chan);
254
255         ast_channel_unsuppress(control->channel, mute_data->direction, mute_data->frametype);
256
257         return NULL;
258 }
259
260 int stasis_app_control_unmute(struct stasis_app_control *control, unsigned int direction, enum ast_frame_type frametype)
261 {
262         struct stasis_app_control_mute_data *mute_data;
263
264         if (!(mute_data = ast_calloc(1, sizeof(*mute_data)))) {
265                 return -1;
266         }
267
268         mute_data->direction = direction;
269         mute_data->frametype = frametype;
270
271         stasis_app_send_command_async(control, app_control_unmute, mute_data);
272
273         return 0;
274 }
275
276 char *stasis_app_control_get_channel_var(struct stasis_app_control *control, const char *variable)
277 {
278         RAII_VAR(struct ast_str *, tmp, ast_str_create(32), ast_free);
279         SCOPED_CHANNELLOCK(lockvar, control->channel);
280
281         if (!tmp) {
282                 return NULL;
283         }
284
285         if (variable[strlen(variable) - 1] == ')') {
286                 if (ast_func_read2(control->channel, variable, &tmp, 0)) {
287                         return NULL;
288                 }
289         } else {
290                 if (!ast_str_retrieve_variable(&tmp, 0, control->channel, NULL, variable)) {
291                         return NULL;
292                 }
293         }
294
295         return ast_strdup(ast_str_buffer(tmp));
296 }
297
298 int stasis_app_control_set_channel_var(struct stasis_app_control *control, const char *variable, const char *value)
299 {
300         return pbx_builtin_setvar_helper(control->channel, variable, value);
301 }
302
303 static void *app_control_hold(struct stasis_app_control *control,
304         struct ast_channel *chan, void *data)
305 {
306         ast_indicate(control->channel, AST_CONTROL_HOLD);
307
308         return NULL;
309 }
310
311 void stasis_app_control_hold(struct stasis_app_control *control)
312 {
313         stasis_app_send_command_async(control, app_control_hold, NULL);
314 }
315
316 static void *app_control_unhold(struct stasis_app_control *control,
317         struct ast_channel *chan, void *data)
318 {
319         ast_indicate(control->channel, AST_CONTROL_UNHOLD);
320
321         return NULL;
322 }
323
324 void stasis_app_control_unhold(struct stasis_app_control *control)
325 {
326         stasis_app_send_command_async(control, app_control_unhold, NULL);
327 }
328
329 static void *app_control_moh_start(struct stasis_app_control *control,
330         struct ast_channel *chan, void *data)
331 {
332         char *moh_class = data;
333
334         ast_moh_start(chan, moh_class, NULL);
335
336         ast_free(moh_class);
337         return NULL;
338 }
339
340 void stasis_app_control_moh_start(struct stasis_app_control *control, const char *moh_class)
341 {
342         char *data = NULL;
343
344         if (!ast_strlen_zero(moh_class)) {
345                 data = ast_strdup(moh_class);
346         }
347
348         stasis_app_send_command_async(control, app_control_moh_start, data);
349 }
350
351 static void *app_control_moh_stop(struct stasis_app_control *control,
352         struct ast_channel *chan, void *data)
353 {
354         ast_moh_stop(chan);
355         return NULL;
356 }
357
358 void stasis_app_control_moh_stop(struct stasis_app_control *control)
359 {
360         stasis_app_send_command_async(control, app_control_moh_stop, NULL);
361 }
362
363 struct ast_channel_snapshot *stasis_app_control_get_snapshot(
364         const struct stasis_app_control *control)
365 {
366         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
367         struct ast_channel_snapshot *snapshot;
368
369         msg = stasis_cache_get(ast_channel_cache(), ast_channel_snapshot_type(),
370                 stasis_app_control_get_channel_id(control));
371         if (!msg) {
372                 return NULL;
373         }
374
375         snapshot = stasis_message_data(msg);
376         ast_assert(snapshot != NULL);
377
378         ao2_ref(snapshot, +1);
379         return snapshot;
380 }
381
382 void *stasis_app_send_command(struct stasis_app_control *control,
383         stasis_app_command_cb command_fn, void *data)
384 {
385         RAII_VAR(struct stasis_app_command *, command, NULL, ao2_cleanup);
386
387         if (control == NULL) {
388                 return NULL;
389         }
390
391         command = exec_command(control, command_fn, data);
392         if (!command) {
393                 return NULL;
394         }
395
396         return command_join(command);
397 }
398
399 int stasis_app_send_command_async(struct stasis_app_control *control,
400         stasis_app_command_cb command_fn, void *data)
401 {
402         RAII_VAR(struct stasis_app_command *, command, NULL, ao2_cleanup);
403
404         if (control == NULL) {
405                 return -1;
406         }
407
408         command = exec_command(control, command_fn, data);
409         if (!command) {
410                 return -1;
411         }
412
413         return 0;
414 }
415
416 const char *stasis_app_control_get_channel_id(
417         const struct stasis_app_control *control)
418 {
419         return ast_channel_uniqueid(control->channel);
420 }
421
422 void stasis_app_control_publish(
423         struct stasis_app_control *control, struct stasis_message *message)
424 {
425         if (!control || !control->channel || !message) {
426                 return;
427         }
428         stasis_publish(ast_channel_topic(control->channel), message);
429 }
430
431 int stasis_app_control_queue_control(struct stasis_app_control *control,
432         enum ast_control_frame_type frame_type)
433 {
434         return ast_queue_control(control->channel, frame_type);
435 }
436
437 int control_dispatch_all(struct stasis_app_control *control,
438         struct ast_channel *chan)
439 {
440         int count = 0;
441         struct ao2_iterator i;
442         void *obj;
443
444         ast_assert(control->channel == chan);
445
446         i = ao2_iterator_init(control->command_queue, AO2_ITERATOR_UNLINK);
447
448         while ((obj = ao2_iterator_next(&i))) {
449                 RAII_VAR(struct stasis_app_command *, command, obj, ao2_cleanup);
450                 command_invoke(command, control, chan);
451                 ++count;
452         }
453
454         ao2_iterator_destroy(&i);
455         return count;
456 }
457
458 /* Must be defined here since it must operate on the channel outside of the queue */
459 int stasis_app_control_remove_channel_from_bridge(
460         struct stasis_app_control *control, struct ast_bridge *bridge)
461 {
462         return ast_bridge_remove(bridge, control->channel);
463 }