70bda4150a1250d5d66ce78daec5558c99c7c657
[asterisk/asterisk.git] / main / message.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2010, Digium, Inc.
5  *
6  * Russell Bryant <russell@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 Out-of-call text message support
22  *
23  * \author Russell Bryant <russell@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/_private.h"
35
36 #include "asterisk/module.h"
37 #include "asterisk/datastore.h"
38 #include "asterisk/pbx.h"
39 #include "asterisk/manager.h"
40 #include "asterisk/strings.h"
41 #include "asterisk/astobj2.h"
42 #include "asterisk/app.h"
43 #include "asterisk/taskprocessor.h"
44 #include "asterisk/message.h"
45
46 /*** DOCUMENTATION
47         <function name="MESSAGE" language="en_US">
48                 <synopsis>
49                         Create a message or read fields from a message.
50                 </synopsis>
51                 <syntax argsep="/">
52                         <parameter name="argument" required="true">
53                         <para>Field of the message to get or set.</para>
54                         <enumlist>
55                                 <enum name="to">
56                                         <para>Read-only.  The destination of the message.  When processing an
57                                         incoming message, this will be set to the destination listed as
58                                         the recipient of the message that was received by Asterisk.</para>
59                                 </enum>
60                                 <enum name="from">
61                                         <para>Read-only.  The source of the message.  When processing an
62                                         incoming message, this will be set to the source of the message.</para>
63                                 </enum>
64                                 <enum name="custom_data">
65                                         <para>Write-only.  Mark or unmark all message headers for an outgoing
66                                         message.  The following values can be set:</para>
67                                         <enumlist>
68                                                 <enum name="mark_all_outbound">
69                                                         <para>Mark all headers for an outgoing message.</para>
70                                                 </enum>
71                                                 <enum name="clear_all_outbound">
72                                                         <para>Unmark all headers for an outgoing message.</para>
73                                                 </enum>
74                                         </enumlist>
75                                 </enum>
76                                 <enum name="body">
77                                         <para>Read/Write.  The message body.  When processing an incoming
78                                         message, this includes the body of the message that Asterisk
79                                         received.  When MessageSend() is executed, the contents of this
80                                         field are used as the body of the outgoing message.  The body
81                                         will always be UTF-8.</para>
82                                 </enum>
83                         </enumlist>
84                         </parameter>
85                 </syntax>
86                 <description>
87                         <para>This function will read from or write a value to a text message.
88                         It is used both to read the data out of an incoming message, as well as
89                         modify or create a message that will be sent outbound.</para>
90                 </description>
91                 <see-also>
92                         <ref type="application">MessageSend</ref>
93                 </see-also>
94         </function>
95         <function name="MESSAGE_DATA" language="en_US">
96                 <synopsis>
97                         Read or write custom data attached to a message.
98                 </synopsis>
99                 <syntax argsep="/">
100                         <parameter name="argument" required="true">
101                         <para>Field of the message to get or set.</para>
102                         </parameter>
103                 </syntax>
104                 <description>
105                         <para>This function will read from or write a value to a text message.
106                         It is used both to read the data out of an incoming message, as well as
107                         modify a message that will be sent outbound.</para>
108                         <note>
109                                 <para>If you want to set an outbound message to carry data in the
110                                 current message, do
111                                 Set(MESSAGE_DATA(<replaceable>key</replaceable>)=${MESSAGE_DATA(<replaceable>key</replaceable>)}).</para>
112                         </note>
113                 </description>
114                 <see-also>
115                         <ref type="application">MessageSend</ref>
116                 </see-also>
117         </function>
118         <application name="MessageSend" language="en_US">
119                 <synopsis>
120                         Send a text message.
121                 </synopsis>
122                 <syntax>
123                         <parameter name="to" required="true">
124                                 <para>A To URI for the message.</para>
125                                 <xi:include xpointer="xpointer(/docs/info[@name='PJSIPMessageToInfo'])" />
126                                 <xi:include xpointer="xpointer(/docs/info[@name='SIPMessageToInfo'])" />
127                                 <xi:include xpointer="xpointer(/docs/info[@name='XMPPMessageToInfo'])" />
128                         </parameter>
129                         <parameter name="from" required="false">
130                                 <para>A From URI for the message if needed for the
131                                 message technology being used to send this message.</para>
132                                 <xi:include xpointer="xpointer(/docs/info[@name='PJSIPMessageFromInfo'])" />
133                                 <xi:include xpointer="xpointer(/docs/info[@name='SIPMessageFromInfo'])" />
134                                 <xi:include xpointer="xpointer(/docs/info[@name='XMPPMessageFromInfo'])" />
135                         </parameter>
136                 </syntax>
137                 <description>
138                         <para>Send a text message.  The body of the message that will be
139                         sent is what is currently set to <literal>MESSAGE(body)</literal>.
140                           The technology chosen for sending the message is determined
141                         based on a prefix to the <literal>to</literal> parameter.</para>
142                         <para>This application sets the following channel variables:</para>
143                         <variablelist>
144                                 <variable name="MESSAGE_SEND_STATUS">
145                                         <para>This is the message delivery status returned by this application.</para>
146                                         <value name="INVALID_PROTOCOL">
147                                                 No handler for the technology part of the URI was found.
148                                         </value>
149                                         <value name="INVALID_URI">
150                                                 The protocol handler reported that the URI was not valid.
151                                         </value>
152                                         <value name="SUCCESS">
153                                                 Successfully passed on to the protocol handler, but delivery has not necessarily been guaranteed.
154                                         </value>
155                                         <value name="FAILURE">
156                                                 The protocol handler reported that it was unabled to deliver the message for some reason.
157                                         </value>
158                                 </variable>
159                         </variablelist>
160                 </description>
161         </application>
162         <manager name="MessageSend" language="en_US">
163                 <synopsis>
164                         Send an out of call message to an endpoint.
165                 </synopsis>
166                 <syntax>
167                         <xi:include xpointer="xpointer(/docs/manager[@name='Login']/syntax/parameter[@name='ActionID'])" />
168                         <parameter name="To" required="true">
169                                 <para>The URI the message is to be sent to.</para>
170                                 <xi:include xpointer="xpointer(/docs/info[@name='PJSIPMessageToInfo'])" />
171                                 <xi:include xpointer="xpointer(/docs/info[@name='SIPMessageToInfo'])" />
172                                 <xi:include xpointer="xpointer(/docs/info[@name='XMPPMessageToInfo'])" />
173                         </parameter>
174                         <parameter name="From">
175                                 <para>A From URI for the message if needed for the
176                                 message technology being used to send this message.</para>
177                                 <xi:include xpointer="xpointer(/docs/info[@name='PJSIPMessageFromInfo'])" />
178                                 <xi:include xpointer="xpointer(/docs/info[@name='SIPMessageFromInfo'])" />
179                                 <xi:include xpointer="xpointer(/docs/info[@name='XMPPMessageFromInfo'])" />
180                         </parameter>
181                         <parameter name="Body">
182                                 <para>The message body text.  This must not contain any newlines as that
183                                 conflicts with the AMI protocol.</para>
184                         </parameter>
185                         <parameter name="Base64Body">
186                                 <para>Text bodies requiring the use of newlines have to be base64 encoded
187                                 in this field.  Base64Body will be decoded before being sent out.
188                                 Base64Body takes precedence over Body.</para>
189                         </parameter>
190                         <parameter name="Variable">
191                                 <para>Message variable to set, multiple Variable: headers are
192                                 allowed.  The header value is a comma separated list of
193                                 name=value pairs.</para>
194                         </parameter>
195                 </syntax>
196         </manager>
197  ***/
198
199 struct msg_data {
200         AST_DECLARE_STRING_FIELDS(
201                 AST_STRING_FIELD(name);
202                 AST_STRING_FIELD(value);
203         );
204         unsigned int send:1; /* Whether to send out on outbound messages */
205 };
206
207 AST_LIST_HEAD_NOLOCK(outhead, msg_data);
208
209 /*!
210  * \brief A message.
211  *
212  * \todo Consider whether stringfields would be an appropriate optimization here.
213  */
214 struct ast_msg {
215         struct ast_str *to;
216         struct ast_str *from;
217         struct ast_str *body;
218         struct ast_str *context;
219         struct ast_str *exten;
220         struct ao2_container *vars;
221 };
222
223 struct ast_msg_tech_holder {
224         const struct ast_msg_tech *tech;
225         /*!
226          * \brief A rwlock for this object
227          *
228          * a read/write lock must be used to protect the wrapper instead
229          * of the ao2 lock. A rdlock must be held to read tech_holder->tech.
230          */
231         ast_rwlock_t tech_lock;
232 };
233
234 static struct ao2_container *msg_techs;
235
236 static struct ast_taskprocessor *msg_q_tp;
237
238 static const char app_msg_send[] = "MessageSend";
239
240 static void msg_ds_destroy(void *data);
241
242 static const struct ast_datastore_info msg_datastore = {
243         .type = "message",
244         .destroy = msg_ds_destroy,
245 };
246
247 static int msg_func_read(struct ast_channel *chan, const char *function,
248                 char *data, char *buf, size_t len);
249 static int msg_func_write(struct ast_channel *chan, const char *function,
250                 char *data, const char *value);
251
252 static struct ast_custom_function msg_function = {
253         .name = "MESSAGE",
254         .read = msg_func_read,
255         .write = msg_func_write,
256 };
257
258 static int msg_data_func_read(struct ast_channel *chan, const char *function,
259                 char *data, char *buf, size_t len);
260 static int msg_data_func_write(struct ast_channel *chan, const char *function,
261                 char *data, const char *value);
262
263 static struct ast_custom_function msg_data_function = {
264         .name = "MESSAGE_DATA",
265         .read = msg_data_func_read,
266         .write = msg_data_func_write,
267 };
268
269 static struct ast_frame *chan_msg_read(struct ast_channel *chan);
270 static int chan_msg_write(struct ast_channel *chan, struct ast_frame *fr);
271 static int chan_msg_indicate(struct ast_channel *chan, int condition,
272                 const void *data, size_t datalen);
273 static int chan_msg_send_digit_begin(struct ast_channel *chan, char digit);
274 static int chan_msg_send_digit_end(struct ast_channel *chan, char digit,
275                 unsigned int duration);
276
277 /*!
278  * \internal
279  * \brief A bare minimum channel technology
280  *
281  * This will not be registered as we never want anything to try
282  * to create Message channels other than internally in this file.
283  */
284 static const struct ast_channel_tech msg_chan_tech_hack = {
285         .type             = "Message",
286         .description      = "Internal Text Message Processing",
287         .read             = chan_msg_read,
288         .write            = chan_msg_write,
289         .indicate         = chan_msg_indicate,
290         .send_digit_begin = chan_msg_send_digit_begin,
291         .send_digit_end   = chan_msg_send_digit_end,
292 };
293
294 /*!
295  * \internal
296  * \brief ast_channel_tech read callback
297  *
298  * This should never be called.  However, we say that about chan_iax2's
299  * read callback, too, and it seems to randomly get called for some
300  * reason.  If it does, a simple NULL frame will suffice.
301  */
302 static struct ast_frame *chan_msg_read(struct ast_channel *chan)
303 {
304         return &ast_null_frame;
305 }
306
307 /*!
308  * \internal
309  * \brief ast_channel_tech write callback
310  *
311  * Throw all frames away.  We don't care about any of them.
312  */
313 static int chan_msg_write(struct ast_channel *chan, struct ast_frame *fr)
314 {
315         return 0;
316 }
317
318 /*!
319  * \internal
320  * \brief ast_channel_tech indicate callback
321  *
322  * The indicate callback is here just so it can return success.
323  * We don't want any callers of ast_indicate() to think something
324  * has failed.  We also don't want ast_indicate() itself to try
325  * to generate inband tones since we didn't tell it that we took
326  * care of it ourselves.
327  */
328 static int chan_msg_indicate(struct ast_channel *chan, int condition,
329                 const void *data, size_t datalen)
330 {
331         return 0;
332 }
333
334 /*!
335  * \internal
336  * \brief ast_channel_tech send_digit_begin callback
337  *
338  * This is here so that just in case a digit comes at a message channel
339  * that the Asterisk core doesn't waste any time trying to generate
340  * inband DTMF in audio.  It's a waste of resources.
341  */
342 static int chan_msg_send_digit_begin(struct ast_channel *chan, char digit)
343 {
344         return 0;
345 }
346
347 /*!
348  * \internal
349  * \brief ast_channel_tech send_digit_end callback
350  *
351  * This is here so that just in case a digit comes at a message channel
352  * that the Asterisk core doesn't waste any time trying to generate
353  * inband DTMF in audio.  It's a waste of resources.
354  */
355 static int chan_msg_send_digit_end(struct ast_channel *chan, char digit,
356                 unsigned int duration)
357 {
358         return 0;
359 }
360
361 static void msg_ds_destroy(void *data)
362 {
363         struct ast_msg *msg = data;
364
365         ao2_ref(msg, -1);
366 }
367
368 static int msg_data_hash_fn(const void *obj, const int flags)
369 {
370         const struct msg_data *data = obj;
371         return ast_str_case_hash(data->name);
372 }
373
374 static int msg_data_cmp_fn(void *obj, void *arg, int flags)
375 {
376         const struct msg_data *one = obj, *two = arg;
377         return !strcasecmp(one->name, two->name) ? CMP_MATCH | CMP_STOP : 0;
378 }
379
380 static void msg_data_destructor(void *obj)
381 {
382         struct msg_data *data = obj;
383         ast_string_field_free_memory(data);
384 }
385
386 static void msg_destructor(void *obj)
387 {
388         struct ast_msg *msg = obj;
389
390         ast_free(msg->to);
391         msg->to = NULL;
392
393         ast_free(msg->from);
394         msg->from = NULL;
395
396         ast_free(msg->body);
397         msg->body = NULL;
398
399         ast_free(msg->context);
400         msg->context = NULL;
401
402         ast_free(msg->exten);
403         msg->exten = NULL;
404
405         ao2_ref(msg->vars, -1);
406 }
407
408 struct ast_msg *ast_msg_alloc(void)
409 {
410         struct ast_msg *msg;
411
412         if (!(msg = ao2_alloc(sizeof(*msg), msg_destructor))) {
413                 return NULL;
414         }
415
416         if (!(msg->to = ast_str_create(32))) {
417                 ao2_ref(msg, -1);
418                 return NULL;
419         }
420
421         if (!(msg->from = ast_str_create(32))) {
422                 ao2_ref(msg, -1);
423                 return NULL;
424         }
425
426         if (!(msg->body = ast_str_create(128))) {
427                 ao2_ref(msg, -1);
428                 return NULL;
429         }
430
431         if (!(msg->context = ast_str_create(16))) {
432                 ao2_ref(msg, -1);
433                 return NULL;
434         }
435
436         if (!(msg->exten = ast_str_create(16))) {
437                 ao2_ref(msg, -1);
438                 return NULL;
439         }
440
441         if (!(msg->vars = ao2_container_alloc(1, msg_data_hash_fn, msg_data_cmp_fn))) {
442                 ao2_ref(msg, -1);
443                 return NULL;
444         }
445
446         ast_str_set(&msg->context, 0, "default");
447
448         return msg;
449 }
450
451 struct ast_msg *ast_msg_ref(struct ast_msg *msg)
452 {
453         ao2_ref(msg, 1);
454         return msg;
455 }
456
457 struct ast_msg *ast_msg_destroy(struct ast_msg *msg)
458 {
459         ao2_ref(msg, -1);
460
461         return NULL;
462 }
463
464 int ast_msg_set_to(struct ast_msg *msg, const char *fmt, ...)
465 {
466         va_list ap;
467         int res;
468
469         va_start(ap, fmt);
470         res = ast_str_set_va(&msg->to, 0, fmt, ap);
471         va_end(ap);
472
473         return res < 0 ? -1 : 0;
474 }
475
476 int ast_msg_set_from(struct ast_msg *msg, const char *fmt, ...)
477 {
478         va_list ap;
479         int res;
480
481         va_start(ap, fmt);
482         res = ast_str_set_va(&msg->from, 0, fmt, ap);
483         va_end(ap);
484
485         return res < 0 ? -1 : 0;
486 }
487
488 int ast_msg_set_body(struct ast_msg *msg, const char *fmt, ...)
489 {
490         va_list ap;
491         int res;
492
493         va_start(ap, fmt);
494         res = ast_str_set_va(&msg->body, 0, fmt, ap);
495         va_end(ap);
496
497         return res < 0 ? -1 : 0;
498 }
499
500 int ast_msg_set_context(struct ast_msg *msg, const char *fmt, ...)
501 {
502         va_list ap;
503         int res;
504
505         va_start(ap, fmt);
506         res = ast_str_set_va(&msg->context, 0, fmt, ap);
507         va_end(ap);
508
509         return res < 0 ? -1 : 0;
510 }
511
512 int ast_msg_set_exten(struct ast_msg *msg, const char *fmt, ...)
513 {
514         va_list ap;
515         int res;
516
517         va_start(ap, fmt);
518         res = ast_str_set_va(&msg->exten, 0, fmt, ap);
519         va_end(ap);
520
521         return res < 0 ? -1 : 0;
522 }
523
524 const char *ast_msg_get_body(const struct ast_msg *msg)
525 {
526         return ast_str_buffer(msg->body);
527 }
528
529 static struct msg_data *msg_data_alloc(void)
530 {
531         struct msg_data *data;
532
533         if (!(data = ao2_alloc(sizeof(*data), msg_data_destructor))) {
534                 return NULL;
535         }
536
537         if (ast_string_field_init(data, 32)) {
538                 ao2_ref(data, -1);
539                 return NULL;
540         }
541
542         return data;
543 }
544
545 static struct msg_data *msg_data_find(struct ao2_container *vars, const char *name)
546 {
547         struct msg_data tmp = {
548                 .name = name,
549         };
550         return ao2_find(vars, &tmp, OBJ_POINTER);
551 }
552
553 static int msg_set_var_full(struct ast_msg *msg, const char *name, const char *value, unsigned int outbound)
554 {
555         struct msg_data *data;
556
557         if (!(data = msg_data_find(msg->vars, name))) {
558                 if (ast_strlen_zero(value)) {
559                         return 0;
560                 }
561                 if (!(data = msg_data_alloc())) {
562                         return -1;
563                 };
564
565                 ast_string_field_set(data, name, name);
566                 ast_string_field_set(data, value, value);
567                 data->send = outbound;
568                 ao2_link(msg->vars, data);
569         } else {
570                 if (ast_strlen_zero(value)) {
571                         ao2_unlink(msg->vars, data);
572                 } else {
573                         ast_string_field_set(data, value, value);
574                         data->send = outbound;
575                 }
576         }
577
578         ao2_ref(data, -1);
579
580         return 0;
581 }
582
583 int ast_msg_set_var_outbound(struct ast_msg *msg, const char *name, const char *value)
584 {
585         return msg_set_var_full(msg, name, value, 1);
586 }
587
588 int ast_msg_set_var(struct ast_msg *msg, const char *name, const char *value)
589 {
590         return msg_set_var_full(msg, name, value, 0);
591 }
592
593 const char *ast_msg_get_var(struct ast_msg *msg, const char *name)
594 {
595         struct msg_data *data;
596         const char *val = NULL;
597
598         if (!(data = msg_data_find(msg->vars, name))) {
599                 return NULL;
600         }
601
602         /* Yep, this definitely looks like val would be a dangling pointer
603          * after the ref count is decremented.  As long as the message structure
604          * is used in a thread safe manner, this will not be the case though.
605          * The ast_msg holds a reference to this object in the msg->vars container. */
606         val = data->value;
607         ao2_ref(data, -1);
608
609         return val;
610 }
611
612 struct ast_msg_var_iterator {
613         struct ao2_iterator i;
614         struct msg_data *current_used;
615 };
616
617 struct ast_msg_var_iterator *ast_msg_var_iterator_init(const struct ast_msg *msg)
618 {
619         struct ast_msg_var_iterator *i;
620         if (!(i = ast_calloc(1, sizeof(*i)))) {
621                 return NULL;
622         }
623
624         i->i = ao2_iterator_init(msg->vars, 0);
625
626         return i;
627 }
628
629 int ast_msg_var_iterator_next(const struct ast_msg *msg, struct ast_msg_var_iterator *i, const char **name, const char **value)
630 {
631         struct msg_data *data;
632
633         /* Skip any that aren't marked for sending out */
634         while ((data = ao2_iterator_next(&i->i)) && !data->send) {
635                 ao2_ref(data, -1);
636         }
637
638         if (!data) {
639                 return 0;
640         }
641
642         if (data->send) {
643                 *name = data->name;
644                 *value = data->value;
645         }
646
647         /* Leave the refcount to be cleaned up by the caller with
648          * ast_msg_var_unref_current after they finish with the pointers to the data */
649         i->current_used = data;
650
651         return 1;
652 }
653
654 void ast_msg_var_unref_current(struct ast_msg_var_iterator *i) {
655         if (i->current_used) {
656                 ao2_ref(i->current_used, -1);
657         }
658         i->current_used = NULL;
659 }
660
661 void ast_msg_var_iterator_destroy(struct ast_msg_var_iterator *i)
662 {
663         ao2_iterator_destroy(&i->i);
664         ast_free(i);
665 }
666
667 static struct ast_channel *create_msg_q_chan(void)
668 {
669         struct ast_channel *chan;
670         struct ast_datastore *ds;
671
672         chan = ast_channel_alloc(1, AST_STATE_UP,
673                         NULL, NULL, NULL,
674                         NULL, NULL, NULL, NULL, 0,
675                         "%s", "Message/ast_msg_queue");
676
677         if (!chan) {
678                 return NULL;
679         }
680
681         ast_channel_tech_set(chan, &msg_chan_tech_hack);
682         ast_channel_unlock(chan);
683         ast_channel_unlink(chan);
684
685         if (!(ds = ast_datastore_alloc(&msg_datastore, NULL))) {
686                 ast_hangup(chan);
687                 return NULL;
688         }
689
690         ast_channel_lock(chan);
691         ast_channel_datastore_add(chan, ds);
692         ast_channel_unlock(chan);
693
694         return chan;
695 }
696
697 /*!
698  * \internal
699  * \brief Run the dialplan for message processing
700  *
701  * \pre The message has already been set up on the msg datastore
702  *      on this channel.
703  */
704 static void msg_route(struct ast_channel *chan, struct ast_msg *msg)
705 {
706         struct ast_pbx_args pbx_args;
707
708         ast_explicit_goto(chan, ast_str_buffer(msg->context), AS_OR(msg->exten, "s"), 1);
709
710         memset(&pbx_args, 0, sizeof(pbx_args));
711         pbx_args.no_hangup_chan = 1,
712         ast_pbx_run_args(chan, &pbx_args);
713 }
714
715 /*!
716  * \internal
717  * \brief Clean up ast_channel after each message
718  *
719  * Reset various bits of state after routing each message so the same ast_channel
720  * can just be reused.
721  */
722 static void chan_cleanup(struct ast_channel *chan)
723 {
724         struct ast_datastore *msg_ds, *ds;
725         struct varshead *headp;
726         struct ast_var_t *vardata;
727
728         ast_channel_lock(chan);
729
730         /*
731          * Remove the msg datastore.  Free its data but keep around the datastore
732          * object and just reuse it.
733          */
734         if ((msg_ds = ast_channel_datastore_find(chan, &msg_datastore, NULL)) && msg_ds->data) {
735                 ast_channel_datastore_remove(chan, msg_ds);
736                 ao2_ref(msg_ds->data, -1);
737                 msg_ds->data = NULL;
738         }
739
740         /*
741          * Destroy all other datastores.
742          */
743         while ((ds = AST_LIST_REMOVE_HEAD(ast_channel_datastores(chan), entry))) {
744                 ast_datastore_free(ds);
745         }
746
747         /*
748          * Destroy all channel variables.
749          */
750         headp = ast_channel_varshead(chan);
751         while ((vardata = AST_LIST_REMOVE_HEAD(headp, entries))) {
752                 ast_var_delete(vardata);
753         }
754
755         /*
756          * Restore msg datastore.
757          */
758         if (msg_ds) {
759                 ast_channel_datastore_add(chan, msg_ds);
760         }
761         /*
762          * Clear softhangup flags.
763          */
764         ast_channel_clear_softhangup(chan, AST_SOFTHANGUP_ALL);
765
766         ast_channel_unlock(chan);
767 }
768
769 static void destroy_msg_q_chan(void *data)
770 {
771         struct ast_channel **chan = data;
772
773         if (!*chan) {
774                 return;
775         }
776
777         ast_channel_release(*chan);
778 }
779
780 AST_THREADSTORAGE_CUSTOM(msg_q_chan, NULL, destroy_msg_q_chan);
781
782 /*!
783  * \internal
784  * \brief Message queue task processor callback
785  *
786  * \retval 0 success
787  * \retval -1 failure
788  *
789  * \note Even though this returns a value, the taskprocessor code ignores the value.
790  */
791 static int msg_q_cb(void *data)
792 {
793         struct ast_msg *msg = data;
794         struct ast_channel **chan_p, *chan;
795         struct ast_datastore *ds;
796
797         if (!(chan_p = ast_threadstorage_get(&msg_q_chan, sizeof(struct ast_channel *)))) {
798                 return -1;
799         }
800         if (!*chan_p) {
801                 if (!(*chan_p = create_msg_q_chan())) {
802                         return -1;
803                 }
804         }
805         chan = *chan_p;
806
807         ast_channel_lock(chan);
808         if (!(ds = ast_channel_datastore_find(chan, &msg_datastore, NULL))) {
809                 ast_channel_unlock(chan);
810                 return -1;
811         }
812         ao2_ref(msg, +1);
813         ds->data = msg;
814         ast_channel_unlock(chan);
815
816         msg_route(chan, msg);
817         chan_cleanup(chan);
818
819         ao2_ref(msg, -1);
820
821         return 0;
822 }
823
824 int ast_msg_queue(struct ast_msg *msg)
825 {
826         int res;
827
828         res = ast_taskprocessor_push(msg_q_tp, msg_q_cb, msg);
829         if (res == -1) {
830                 ao2_ref(msg, -1);
831         }
832
833         return res;
834 }
835
836 /*!
837  * \internal
838  * \brief Find or create a message datastore on a channel
839  *
840  * \pre chan is locked
841  *
842  * \param chan the relevant channel
843  *
844  * \return the channel's message datastore, or NULL on error
845  */
846 static struct ast_datastore *msg_datastore_find_or_create(struct ast_channel *chan)
847 {
848         struct ast_datastore *ds;
849
850         if ((ds = ast_channel_datastore_find(chan, &msg_datastore, NULL))) {
851                 return ds;
852         }
853
854         if (!(ds = ast_datastore_alloc(&msg_datastore, NULL))) {
855                 return NULL;
856         }
857
858         if (!(ds->data = ast_msg_alloc())) {
859                 ast_datastore_free(ds);
860                 return NULL;
861         }
862
863         ast_channel_datastore_add(chan, ds);
864
865         return ds;
866 }
867
868 static int msg_func_read(struct ast_channel *chan, const char *function,
869                 char *data, char *buf, size_t len)
870 {
871         struct ast_datastore *ds;
872         struct ast_msg *msg;
873
874         ast_channel_lock(chan);
875
876         if (!(ds = ast_channel_datastore_find(chan, &msg_datastore, NULL))) {
877                 ast_channel_unlock(chan);
878                 ast_log(LOG_ERROR, "No MESSAGE data found on the channel to read.\n");
879                 return -1;
880         }
881
882         msg = ds->data;
883         ao2_ref(msg, +1);
884         ast_channel_unlock(chan);
885
886         ao2_lock(msg);
887
888         if (!strcasecmp(data, "to")) {
889                 ast_copy_string(buf, ast_str_buffer(msg->to), len);
890         } else if (!strcasecmp(data, "from")) {
891                 ast_copy_string(buf, ast_str_buffer(msg->from), len);
892         } else if (!strcasecmp(data, "body")) {
893                 ast_copy_string(buf, ast_msg_get_body(msg), len);
894         } else {
895                 ast_log(LOG_WARNING, "Invalid argument to MESSAGE(): '%s'\n", data);
896         }
897
898         ao2_unlock(msg);
899         ao2_ref(msg, -1);
900
901         return 0;
902 }
903
904 static int msg_func_write(struct ast_channel *chan, const char *function,
905                 char *data, const char *value)
906 {
907         struct ast_datastore *ds;
908         struct ast_msg *msg;
909
910         ast_channel_lock(chan);
911
912         if (!(ds = msg_datastore_find_or_create(chan))) {
913                 ast_channel_unlock(chan);
914                 return -1;
915         }
916
917         msg = ds->data;
918         ao2_ref(msg, +1);
919         ast_channel_unlock(chan);
920
921         ao2_lock(msg);
922
923         if (!strcasecmp(data, "to")) {
924                 ast_msg_set_to(msg, "%s", value);
925         } else if (!strcasecmp(data, "from")) {
926                 ast_msg_set_from(msg, "%s", value);
927         } else if (!strcasecmp(data, "body")) {
928                 ast_msg_set_body(msg, "%s", value);
929         } else if (!strcasecmp(data, "custom_data")) {
930                 int outbound = -1;
931                 if (!strcasecmp(value, "mark_all_outbound")) {
932                         outbound = 1;
933                 } else if (!strcasecmp(value, "clear_all_outbound")) {
934                         outbound = 0;
935                 } else {
936                         ast_log(LOG_WARNING, "'%s' is not a valid value for custom_data\n", value);
937                 }
938
939                 if (outbound != -1) {
940                         struct msg_data *hdr_data;
941                         struct ao2_iterator iter = ao2_iterator_init(msg->vars, 0);
942
943                         while ((hdr_data = ao2_iterator_next(&iter))) {
944                                 hdr_data->send = outbound;
945                                 ao2_ref(hdr_data, -1);
946                         }
947                         ao2_iterator_destroy(&iter);
948                 }
949         } else {
950                 ast_log(LOG_WARNING, "'%s' is not a valid write argument.\n", data);
951         }
952
953         ao2_unlock(msg);
954         ao2_ref(msg, -1);
955
956         return 0;
957 }
958
959 static int msg_data_func_read(struct ast_channel *chan, const char *function,
960                 char *data, char *buf, size_t len)
961 {
962         struct ast_datastore *ds;
963         struct ast_msg *msg;
964         const char *val;
965
966         ast_channel_lock(chan);
967
968         if (!(ds = ast_channel_datastore_find(chan, &msg_datastore, NULL))) {
969                 ast_channel_unlock(chan);
970                 ast_log(LOG_ERROR, "No MESSAGE data found on the channel to read.\n");
971                 return -1;
972         }
973
974         msg = ds->data;
975         ao2_ref(msg, +1);
976         ast_channel_unlock(chan);
977
978         ao2_lock(msg);
979
980         if ((val = ast_msg_get_var(msg, data))) {
981                 ast_copy_string(buf, val, len);
982         }
983
984         ao2_unlock(msg);
985         ao2_ref(msg, -1);
986
987         return 0;
988 }
989
990 static int msg_data_func_write(struct ast_channel *chan, const char *function,
991                 char *data, const char *value)
992 {
993         struct ast_datastore *ds;
994         struct ast_msg *msg;
995
996         ast_channel_lock(chan);
997
998         if (!(ds = msg_datastore_find_or_create(chan))) {
999                 ast_channel_unlock(chan);
1000                 return -1;
1001         }
1002
1003         msg = ds->data;
1004         ao2_ref(msg, +1);
1005         ast_channel_unlock(chan);
1006
1007         ao2_lock(msg);
1008
1009         ast_msg_set_var_outbound(msg, data, value);
1010
1011         ao2_unlock(msg);
1012         ao2_ref(msg, -1);
1013
1014         return 0;
1015 }
1016 static int msg_tech_hash(const void *obj, const int flags)
1017 {
1018         struct ast_msg_tech_holder *tech_holder = (struct ast_msg_tech_holder *) obj;
1019         int res = 0;
1020
1021         ast_rwlock_rdlock(&tech_holder->tech_lock);
1022         if (tech_holder->tech) {
1023                 res = ast_str_case_hash(tech_holder->tech->name);
1024         }
1025         ast_rwlock_unlock(&tech_holder->tech_lock);
1026
1027         return res;
1028 }
1029
1030 static int msg_tech_cmp(void *obj, void *arg, int flags)
1031 {
1032         struct ast_msg_tech_holder *tech_holder = obj;
1033         const struct ast_msg_tech_holder *tech_holder2 = arg;
1034         int res = 1;
1035
1036         ast_rwlock_rdlock(&tech_holder->tech_lock);
1037         /*
1038          * tech_holder2 is a temporary fake tech_holder.
1039          */
1040         if (tech_holder->tech) {
1041                 res = strcasecmp(tech_holder->tech->name, tech_holder2->tech->name) ? 0 : CMP_MATCH | CMP_STOP;
1042         }
1043         ast_rwlock_unlock(&tech_holder->tech_lock);
1044
1045         return res;
1046 }
1047
1048 static struct ast_msg_tech_holder *msg_find_by_tech(const struct ast_msg_tech *msg_tech, int ao2_flags)
1049 {
1050         struct ast_msg_tech_holder *tech_holder;
1051         struct ast_msg_tech_holder tmp_tech_holder = {
1052                 .tech = msg_tech,
1053         };
1054
1055         ast_rwlock_init(&tmp_tech_holder.tech_lock);
1056         tech_holder = ao2_find(msg_techs, &tmp_tech_holder, ao2_flags);
1057         ast_rwlock_destroy(&tmp_tech_holder.tech_lock);
1058         return tech_holder;
1059 }
1060
1061 static struct ast_msg_tech_holder *msg_find_by_tech_name(const char *tech_name, int ao2_flags)
1062 {
1063         struct ast_msg_tech tmp_msg_tech = {
1064                 .name = tech_name,
1065         };
1066         return msg_find_by_tech(&tmp_msg_tech, ao2_flags);
1067 }
1068
1069 /*!
1070  * \internal
1071  * \brief MessageSend() application
1072  */
1073 static int msg_send_exec(struct ast_channel *chan, const char *data)
1074 {
1075         struct ast_datastore *ds;
1076         struct ast_msg *msg;
1077         char *tech_name;
1078         struct ast_msg_tech_holder *tech_holder = NULL;
1079         char *parse;
1080         int res = -1;
1081         AST_DECLARE_APP_ARGS(args,
1082                 AST_APP_ARG(to);
1083                 AST_APP_ARG(from);
1084         );
1085
1086         if (ast_strlen_zero(data)) {
1087                 ast_log(LOG_WARNING, "An argument is required to MessageSend()\n");
1088                 pbx_builtin_setvar_helper(chan, "MESSAGE_SEND_STATUS", "INVALID_URI");
1089                 return 0;
1090         }
1091
1092         parse = ast_strdupa(data);
1093         AST_STANDARD_APP_ARGS(args, parse);
1094
1095         if (ast_strlen_zero(args.to)) {
1096                 ast_log(LOG_WARNING, "A 'to' URI is required for MessageSend()\n");
1097                 pbx_builtin_setvar_helper(chan, "MESSAGE_SEND_STATUS", "INVALID_URI");
1098                 return 0;
1099         }
1100
1101         ast_channel_lock(chan);
1102
1103         if (!(ds = ast_channel_datastore_find(chan, &msg_datastore, NULL))) {
1104                 ast_channel_unlock(chan);
1105                 ast_log(LOG_WARNING, "No message data found on channel to send.\n");
1106                 pbx_builtin_setvar_helper(chan, "MESSAGE_SEND_STATUS", "FAILURE");
1107                 return 0;
1108         }
1109
1110         msg = ds->data;
1111         ao2_ref(msg, +1);
1112         ast_channel_unlock(chan);
1113
1114         tech_name = ast_strdupa(args.to);
1115         tech_name = strsep(&tech_name, ":");
1116
1117         tech_holder = msg_find_by_tech_name(tech_name, OBJ_POINTER);
1118
1119         if (!tech_holder) {
1120                 ast_log(LOG_WARNING, "No message technology '%s' found.\n", tech_name);
1121                 pbx_builtin_setvar_helper(chan, "MESSAGE_SEND_STATUS", "INVALID_PROTOCOL");
1122                 goto exit_cleanup;
1123         }
1124
1125         /*
1126          * The message lock is held here to safely allow the technology
1127          * implementation to access the message fields without worrying
1128          * that they could change.
1129          */
1130         ao2_lock(msg);
1131         ast_rwlock_rdlock(&tech_holder->tech_lock);
1132         if (tech_holder->tech) {
1133                 res = tech_holder->tech->msg_send(msg, S_OR(args.to, ""),
1134                                                         S_OR(args.from, ""));
1135         }
1136         ast_rwlock_unlock(&tech_holder->tech_lock);
1137         ao2_unlock(msg);
1138
1139         pbx_builtin_setvar_helper(chan, "MESSAGE_SEND_STATUS", res ? "FAILURE" : "SUCCESS");
1140
1141 exit_cleanup:
1142         if (tech_holder) {
1143                 ao2_ref(tech_holder, -1);
1144                 tech_holder = NULL;
1145         }
1146
1147         ao2_ref(msg, -1);
1148
1149         return 0;
1150 }
1151
1152 static int action_messagesend(struct mansession *s, const struct message *m)
1153 {
1154         const char *to = ast_strdupa(astman_get_header(m, "To"));
1155         const char *from = astman_get_header(m, "From");
1156         const char *body = astman_get_header(m, "Body");
1157         const char *base64body = astman_get_header(m, "Base64Body");
1158         char base64decoded[1301] = { 0, };
1159         char *tech_name = NULL;
1160         struct ast_variable *vars = NULL;
1161         struct ast_variable *data = NULL;
1162         struct ast_msg_tech_holder *tech_holder = NULL;
1163         struct ast_msg *msg;
1164         int res = -1;
1165
1166         if (ast_strlen_zero(to)) {
1167                 astman_send_error(s, m, "No 'To' address specified.");
1168                 return -1;
1169         }
1170
1171         if (!ast_strlen_zero(base64body)) {
1172                 ast_base64decode((unsigned char *) base64decoded, base64body, sizeof(base64decoded) - 1);
1173                 body = base64decoded;
1174         }
1175
1176         tech_name = ast_strdupa(to);
1177         tech_name = strsep(&tech_name, ":");
1178
1179         tech_holder = msg_find_by_tech_name(tech_name, OBJ_POINTER);
1180
1181         if (!tech_holder) {
1182                 astman_send_error(s, m, "Message technology not found.");
1183                 return -1;
1184         }
1185
1186         if (!(msg = ast_msg_alloc())) {
1187                 ao2_ref(tech_holder, -1);
1188                 astman_send_error(s, m, "Internal failure\n");
1189                 return -1;
1190         }
1191
1192         data = astman_get_variables(m);
1193         for (vars = data; vars; vars = vars->next) {
1194                 ast_msg_set_var_outbound(msg, vars->name, vars->value);
1195         }
1196
1197         ast_msg_set_body(msg, "%s", body);
1198
1199         ast_rwlock_rdlock(&tech_holder->tech_lock);
1200         if (tech_holder->tech) {
1201                 res = tech_holder->tech->msg_send(msg, S_OR(to, ""), S_OR(from, ""));
1202         }
1203         ast_rwlock_unlock(&tech_holder->tech_lock);
1204
1205         ast_variables_destroy(vars);
1206         ao2_ref(tech_holder, -1);
1207         ao2_ref(msg, -1);
1208
1209         if (res) {
1210                 astman_send_error(s, m, "Message failed to send.");
1211         } else {
1212                 astman_send_ack(s, m, "Message successfully sent");
1213         }
1214         return res;
1215 }
1216
1217 int ast_msg_send(struct ast_msg *msg, const char *to, const char *from)
1218 {
1219         char *tech_name = NULL;
1220         struct ast_msg_tech_holder *tech_holder = NULL;
1221         int res = -1;
1222
1223         if (ast_strlen_zero(to)) {
1224                 ao2_ref(msg, -1);
1225                 return -1;
1226         }
1227
1228         tech_name = ast_strdupa(to);
1229         tech_name = strsep(&tech_name, ":");
1230
1231         tech_holder = msg_find_by_tech_name(tech_name, OBJ_POINTER);
1232
1233         if (!tech_holder) {
1234                 ao2_ref(msg, -1);
1235                 return -1;
1236         }
1237
1238         ast_rwlock_rdlock(&tech_holder->tech_lock);
1239         if (tech_holder->tech) {
1240                 res = tech_holder->tech->msg_send(msg, S_OR(to, ""), S_OR(from, ""));
1241         }
1242         ast_rwlock_unlock(&tech_holder->tech_lock);
1243
1244         ao2_ref(tech_holder, -1);
1245         ao2_ref(msg, -1);
1246
1247         return res;
1248 }
1249
1250 int ast_msg_tech_register(const struct ast_msg_tech *tech)
1251 {
1252         struct ast_msg_tech_holder *tech_holder;
1253
1254         if ((tech_holder = msg_find_by_tech(tech, OBJ_POINTER))) {
1255                 ao2_ref(tech_holder, -1);
1256                 ast_log(LOG_ERROR, "Message technology already registered for '%s'\n",
1257                                 tech->name);
1258                 return -1;
1259         }
1260
1261         if (!(tech_holder = ao2_alloc(sizeof(*tech_holder), NULL))) {
1262                 return -1;
1263         }
1264
1265         ast_rwlock_init(&tech_holder->tech_lock);
1266         tech_holder->tech = tech;
1267
1268         ao2_link(msg_techs, tech_holder);
1269
1270         ao2_ref(tech_holder, -1);
1271         tech_holder = NULL;
1272
1273         ast_verb(3, "Message technology handler '%s' registered.\n", tech->name);
1274
1275         return 0;
1276 }
1277
1278 int ast_msg_tech_unregister(const struct ast_msg_tech *tech)
1279 {
1280         struct ast_msg_tech_holder *tech_holder;
1281
1282         tech_holder = msg_find_by_tech(tech, OBJ_POINTER | OBJ_UNLINK);
1283
1284         if (!tech_holder) {
1285                 ast_log(LOG_ERROR, "No '%s' message technology found.\n", tech->name);
1286                 return -1;
1287         }
1288
1289         ast_rwlock_wrlock(&tech_holder->tech_lock);
1290         tech_holder->tech = NULL;
1291         ast_rwlock_unlock(&tech_holder->tech_lock);
1292
1293         ao2_ref(tech_holder, -1);
1294         tech_holder = NULL;
1295
1296         ast_verb(3, "Message technology handler '%s' unregistered.\n", tech->name);
1297
1298         return 0;
1299 }
1300
1301 void ast_msg_shutdown(void)
1302 {
1303         if (msg_q_tp) {
1304                 msg_q_tp = ast_taskprocessor_unreference(msg_q_tp);
1305         }
1306 }
1307
1308 /*!
1309  * \internal
1310  * \brief Clean up other resources on Asterisk shutdown
1311  *
1312  * \note This does not include the msg_q_tp object, which must be disposed
1313  * of prior to Asterisk checking for channel destruction in its shutdown
1314  * sequence.  The atexit handlers are executed after this occurs.
1315  */
1316 static void message_shutdown(void)
1317 {
1318         ast_custom_function_unregister(&msg_function);
1319         ast_custom_function_unregister(&msg_data_function);
1320         ast_unregister_application(app_msg_send);
1321         ast_manager_unregister("MessageSend");
1322
1323         if (msg_techs) {
1324                 ao2_ref(msg_techs, -1);
1325                 msg_techs = NULL;
1326         }
1327 }
1328
1329 /*
1330  * \internal
1331  * \brief Initialize stuff during Asterisk startup.
1332  *
1333  * Cleanup isn't a big deal in this function.  If we return non-zero,
1334  * Asterisk is going to exit.
1335  *
1336  * \retval 0 success
1337  * \retval non-zero failure
1338  */
1339 int ast_msg_init(void)
1340 {
1341         int res;
1342
1343         msg_q_tp = ast_taskprocessor_get("ast_msg_queue", TPS_REF_DEFAULT);
1344         if (!msg_q_tp) {
1345                 return -1;
1346         }
1347
1348         msg_techs = ao2_container_alloc(17, msg_tech_hash, msg_tech_cmp);
1349         if (!msg_techs) {
1350                 return -1;
1351         }
1352
1353         res = __ast_custom_function_register(&msg_function, NULL);
1354         res |= __ast_custom_function_register(&msg_data_function, NULL);
1355         res |= ast_register_application2(app_msg_send, msg_send_exec, NULL, NULL, NULL);
1356         res |= ast_manager_register_xml_core("MessageSend", EVENT_FLAG_MESSAGE, action_messagesend);
1357
1358         ast_register_atexit(message_shutdown);
1359
1360         return res;
1361 }