Migrate a large number of AMI events over to Stasis-Core
[asterisk/asterisk.git] / res / res_sip_mwi.c
1 /*
2  * Asterisk -- An open source telephony toolkit.
3  *
4  * Copyright (C) 2013, Digium, Inc.
5  *
6  * Mark Michelson <mmichelson@digium.com>
7  *
8  * See http://www.asterisk.org for more information about
9  * the Asterisk project. Please do not directly contact
10  * any of the maintainers of this project for assistance;
11  * the project provides a web site, mailing lists and IRC
12  * channels for your use.
13  *
14  * This program is free software, distributed under the terms of
15  * the GNU General Public License Version 2. See the LICENSE file
16  * at the top of the source tree.
17  */
18
19 /*** MODULEINFO
20         <depend>pjproject</depend>
21         <depend>res_sip</depend>
22         <support_level>core</support_level>
23  ***/
24
25 #include "asterisk.h"
26
27 #include <pjsip.h>
28 #include <pjsip_simple.h>
29 #include <pjlib.h>
30
31 #include "asterisk/res_sip.h"
32 #include "asterisk/res_sip_pubsub.h"
33 #include "asterisk/module.h"
34 #include "asterisk/logger.h"
35 #include "asterisk/astobj2.h"
36 #include "asterisk/sorcery.h"
37 #include "asterisk/stasis.h"
38 #include "asterisk/app.h"
39
40 /*** MODULEINFO
41         <depend>pjproject</depend>
42         <depend>res_sip</depend>
43         <depend>res_sip_pubsub</depend>
44         <support_level>core</support_level>
45  ***/
46
47 struct mwi_subscription;
48 AO2_GLOBAL_OBJ_STATIC(unsolicited_mwi);
49
50 #define STASIS_BUCKETS 13
51 #define MWI_BUCKETS 53
52 static void mwi_subscription_shutdown(struct ast_sip_subscription *sub);
53 static struct ast_sip_subscription *mwi_new_subscribe(struct ast_sip_endpoint *endpoint,
54                 pjsip_rx_data *rdata);
55 static void mwi_resubscribe(struct ast_sip_subscription *sub, pjsip_rx_data *rdata,
56                 struct ast_sip_subscription_response_data *response_data);
57 static void mwi_subscription_timeout(struct ast_sip_subscription *sub);
58 static void mwi_subscription_terminated(struct ast_sip_subscription *sub, pjsip_rx_data *rdata);
59 static void mwi_notify_response(struct ast_sip_subscription *sub, pjsip_rx_data *rdata);
60 static void mwi_notify_request(struct ast_sip_subscription *sub, pjsip_rx_data *rdata,
61                 struct ast_sip_subscription_response_data *response_data);
62 static int mwi_refresh_subscription(struct ast_sip_subscription *sub);
63
64 static struct ast_sip_subscription_handler mwi_handler = {
65         .event_name = "message-summary",
66         .accept = { "application/simple-message-summary", },
67         .subscription_shutdown = mwi_subscription_shutdown,
68         .new_subscribe = mwi_new_subscribe,
69         .resubscribe = mwi_resubscribe,
70         .subscription_timeout = mwi_subscription_timeout,
71         .subscription_terminated = mwi_subscription_terminated,
72         .notify_response = mwi_notify_response,
73         .notify_request = mwi_notify_request,
74         .refresh_subscription = mwi_refresh_subscription,
75 };
76
77 /*!
78  * \brief Wrapper for stasis subscription
79  *
80  * An MWI subscription has a container of these. This
81  * represents a stasis subscription for MWI state.
82  */
83 struct mwi_stasis_subscription {
84         /*! The MWI stasis subscription */
85         struct stasis_subscription *stasis_sub;
86         /*! The mailbox corresponding with the MWI subscription. Used as a hash key */
87         char mailbox[1];
88 };
89
90 /*!
91  * \brief A subscription for MWI
92  *
93  * This subscription is the basis for MWI for an endpoint. Each
94  * endpoint that uses MWI will have a corresponding mwi_subscription.
95  *
96  * This structure acts as the owner for the underlying SIP subscription.
97  * When the mwi_subscription is destroyed, the SIP subscription dies, too.
98  * The mwi_subscription's lifetime is governed by its underlying stasis
99  * subscriptions. When all stasis subscriptions are destroyed, the
100  * mwi_subscription is destroyed as well.
101  */
102 struct mwi_subscription {
103         /*! Container of \ref mwi_stasis_subscription structures.
104          * A single MWI subscription may be fore multiple mailboxes, thus
105          * requiring multiple stasis subscriptions
106          */
107         struct ao2_container *stasis_subs;
108         /*! The SIP subscription. Unsolicited MWI does not use this */
109         struct ast_sip_subscription *sip_sub;
110         /*! Is the MWI solicited (i.e. Initiated with an external SUBSCRIBE) ? */
111         unsigned int is_solicited;
112         /*! Identifier for the subscription.
113          * The identifier is the same as the corresponding endpoint's stasis ID.
114          * Used as a hash key
115          */
116         char id[1];
117 };
118
119 static void mwi_stasis_cb(void *userdata, struct stasis_subscription *sub,
120                 struct stasis_topic *topic, struct stasis_message *msg);
121
122 static struct mwi_stasis_subscription *mwi_stasis_subscription_alloc(const char *mailbox, struct mwi_subscription *mwi_sub)
123 {
124         struct mwi_stasis_subscription *mwi_stasis_sub;
125         struct stasis_topic *topic;
126         
127         if (!mwi_sub) {
128                 return NULL;
129         }
130         
131         mwi_stasis_sub = ao2_alloc(sizeof(*mwi_stasis_sub) + strlen(mailbox), NULL);
132         if (!mwi_stasis_sub) {
133                 return NULL;
134         }
135
136         topic = ast_mwi_topic(mailbox);
137
138         /* Safe strcpy */
139         strcpy(mwi_stasis_sub->mailbox, mailbox);
140         ao2_ref(mwi_sub, +1);
141         ast_debug(3, "Creating stasis MWI subscription to mailbox %s for endpoint %s\n", mailbox, mwi_sub->id);
142         mwi_stasis_sub->stasis_sub = stasis_subscribe(topic, mwi_stasis_cb, mwi_sub);
143         return mwi_stasis_sub;
144 }
145
146 static int stasis_sub_hash(const void *obj, int flags)
147 {
148         const struct mwi_stasis_subscription *mwi_stasis = obj;
149
150         return ast_str_hash(mwi_stasis->mailbox);
151 }
152
153 static int stasis_sub_cmp(void *obj, void *arg, int flags)
154 {
155         struct mwi_stasis_subscription *mwi_stasis1 = obj;
156         struct mwi_stasis_subscription *mwi_stasis2 = arg;
157
158         return strcmp(mwi_stasis1->mailbox, mwi_stasis2->mailbox) ? 0 : CMP_MATCH;
159 }
160
161 static void mwi_subscription_destructor(void *obj)
162 {
163         struct mwi_subscription *sub = obj;
164
165         ast_debug(3, "Destroying MWI subscription for endpoint %s\n", sub->id);
166         ao2_cleanup(sub->sip_sub);
167         ao2_cleanup(sub->stasis_subs);
168 }
169
170 static struct mwi_subscription *mwi_subscription_alloc(struct ast_sip_endpoint *endpoint,
171                 enum ast_sip_subscription_role role, unsigned int is_solicited, pjsip_rx_data *rdata)
172 {
173         struct mwi_subscription *sub;
174         const char *endpoint_id = ast_sorcery_object_get_id(endpoint);
175
176         sub = ao2_alloc(sizeof(*sub) + strlen(endpoint_id),
177                         mwi_subscription_destructor);
178
179         if (!sub) {
180                 return NULL;
181         }
182
183         /* Safe strcpy */
184         strcpy(sub->id, endpoint_id);
185         /* Unsolicited MWI doesn't actually result in a SIP subscription being
186          * created. This is because a SIP subscription associates with a dialog.
187          * Most devices expect unsolicited MWI NOTIFYs to appear out of dialog. If
188          * they receive an in-dialog MWI NOTIFY (i.e. with a to-tag), then they
189          * will reject the NOTIFY with a 481, thus resulting in message-waiting
190          * state not being updated on the device
191          */
192         if (is_solicited) {
193                 sub->sip_sub = ast_sip_create_subscription(&mwi_handler,
194                                 role, endpoint, rdata);
195                 if (!sub->sip_sub) {
196                         ast_log(LOG_WARNING, "Unable to create MWI SIP subscription for endpoint %s\n", sub->id);
197                         ao2_cleanup(sub);
198                         return NULL;
199                 }
200         }
201
202         sub->stasis_subs = ao2_container_alloc(STASIS_BUCKETS, stasis_sub_hash, stasis_sub_cmp);
203         if (!sub->stasis_subs) {
204                 ao2_cleanup(sub);
205                 return NULL;
206         }
207         sub->is_solicited = is_solicited;
208
209         ast_debug(3, "Created %s MWI subscription for endpoint %s\n", is_solicited ? "solicited" : "unsolicited", sub->id);
210
211         return sub;
212 }
213
214 static int mwi_sub_hash(const void *obj, int flags)
215 {
216         const struct mwi_subscription *mwi_sub = obj;
217         
218         return ast_str_hash(mwi_sub->id);
219 }
220
221 static int mwi_sub_cmp(void *obj, void *arg, int flags)
222 {
223         struct mwi_subscription *mwi_sub1 = obj;
224         struct mwi_subscription *mwi_sub2 = arg;
225
226         return strcmp(mwi_sub1->id, mwi_sub2->id) ? 0 : CMP_MATCH;
227 }
228
229 struct message_accumulator {
230         int old_msgs;
231         int new_msgs;
232         const char *reason;
233 };
234
235 static int get_message_count(void *obj, void *arg, int flags)
236 {
237         RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
238         struct mwi_stasis_subscription *mwi_stasis = obj;
239         struct message_accumulator *counter = arg;
240         struct ast_mwi_state *mwi_state;
241
242         msg = stasis_cache_get(ast_mwi_topic_cached(), ast_mwi_state_type(), mwi_stasis->mailbox);
243         if (!msg) {
244                 return 0;
245         }
246
247         mwi_state = stasis_message_data(msg);
248         counter->old_msgs += mwi_state->old_msgs;
249         counter->new_msgs += mwi_state->new_msgs;
250         return 0;
251 }
252
253 struct unsolicited_mwi_data {
254         struct mwi_subscription *sub;
255         struct ast_sip_endpoint *endpoint;
256         pjsip_evsub_state state;
257         const char *reason;
258         const pjsip_media_type *mwi_type;
259         const pj_str_t *body_text;
260 };
261
262 static int send_unsolicited_mwi_notify_to_contact(void *obj, void *arg, int flags)
263 {
264         struct unsolicited_mwi_data *mwi_data = arg;
265         struct mwi_subscription *sub = mwi_data->sub;
266         struct ast_sip_endpoint *endpoint = mwi_data->endpoint;
267         pjsip_evsub_state state = mwi_data->state;
268         const char *reason = mwi_data->reason;
269         const pjsip_media_type *mwi_type = mwi_data->mwi_type;
270         const pj_str_t *body_text = mwi_data->body_text;
271         struct ast_sip_contact *contact = obj;
272         const char *state_name;
273         pjsip_tx_data *tdata;
274         pjsip_msg_body *msg_body;
275         pjsip_sub_state_hdr *sub_state;
276         pjsip_event_hdr *event;
277         const pjsip_hdr *allow_events = pjsip_evsub_get_allow_events_hdr(NULL);
278
279         if (ast_sip_create_request("NOTIFY", NULL, endpoint, contact->uri, &tdata)) {
280                 ast_log(LOG_WARNING, "Unable to create unsolicited NOTIFY request to endpoint %s URI %s\n", sub->id, contact->uri);
281                 return 0;
282         }
283
284         switch (state) {
285         case PJSIP_EVSUB_STATE_ACTIVE:
286                 state_name = "active";
287                 break;
288         case PJSIP_EVSUB_STATE_TERMINATED:
289         default:
290                 state_name = "terminated";
291                 break;
292         }
293
294         sub_state = pjsip_sub_state_hdr_create(tdata->pool);
295         pj_cstr(&sub_state->sub_state, state_name);
296         if (reason) {
297                 pj_cstr(&sub_state->reason_param, reason);
298         }
299         pjsip_msg_add_hdr(tdata->msg, (pjsip_hdr *) sub_state);
300
301         event = pjsip_event_hdr_create(tdata->pool);
302         pj_cstr(&event->event_type, "message-summary");
303         pjsip_msg_add_hdr(tdata->msg, (pjsip_hdr *) event);
304
305         pjsip_msg_add_hdr(tdata->msg, pjsip_hdr_shallow_clone(tdata->pool, allow_events));
306         msg_body = pjsip_msg_body_create(tdata->pool, &mwi_type->type, &mwi_type->subtype, body_text);
307         tdata->msg->body = msg_body;
308         ast_sip_send_request(tdata, NULL, endpoint);
309
310         return 0;
311 }
312
313 static void send_unsolicited_mwi_notify(struct mwi_subscription *sub, pjsip_evsub_state state, const char *reason,
314                 const pjsip_media_type *mwi_type, const pj_str_t *body_text)
315 {
316         RAII_VAR(struct ast_sip_endpoint *, endpoint, ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(),
317                                 "endpoint", sub->id), ao2_cleanup);
318         char *endpoint_aors;
319         char *aor_name;
320
321         if (!endpoint) {
322                 ast_log(LOG_WARNING, "Unable to send unsolicited MWI to %s because endpoint does not exist\n",
323                                 sub->id);
324                 return;
325         }
326         if (ast_strlen_zero(endpoint->aors)) {
327                 ast_log(LOG_WARNING, "Unable to send unsolicited MWI to %s because the endpoint has no"
328                                 " configured AORs\n", sub->id);
329                 return;
330         }
331
332         endpoint_aors = ast_strdupa(endpoint->aors);
333
334         while ((aor_name = strsep(&endpoint_aors, ","))) {
335                 RAII_VAR(struct ast_sip_aor *, aor, ast_sip_location_retrieve_aor(aor_name), ao2_cleanup);
336                 RAII_VAR(struct ao2_container *, contacts, NULL, ao2_cleanup);
337                 struct unsolicited_mwi_data mwi_data = {
338                         .sub = sub,
339                         .endpoint = endpoint,
340                         .state = state,
341                         .reason = reason,
342                         .mwi_type = mwi_type,
343                         .body_text = body_text,
344                 };
345
346                 if (!aor) {
347                         ast_log(LOG_WARNING, "Unable to locate AOR %s for unsolicited MWI\n", aor_name);
348                         continue;
349                 }
350
351                 contacts = ast_sip_location_retrieve_aor_contacts(aor);
352                 if (!contacts || (ao2_container_count(contacts) == 0)) {
353                         ast_log(LOG_WARNING, "No contacts bound to AOR %s. Cannot send unsolicited MWI.\n", aor_name);
354                         continue;
355                 }
356
357                 ao2_callback(contacts, OBJ_NODATA, send_unsolicited_mwi_notify_to_contact, &mwi_data);
358         }
359 }
360
361 static void send_mwi_notify(struct mwi_subscription *sub, pjsip_evsub_state state, const char *reason)
362 {
363         const pj_str_t *reason_str_ptr = NULL;
364         static pjsip_media_type mwi_type = {
365                 .type = { "application", 11 },
366                 .subtype = { "simple-message-summary", 22 },
367         };
368         struct message_accumulator counter = {
369                 .old_msgs = 0,
370                 .new_msgs = 0,
371         };
372         RAII_VAR(struct ast_str *, body, ast_str_create(64), ast_free_ptr);
373         pjsip_tx_data *tdata;
374         pj_str_t reason_str;
375         pj_str_t pj_body;
376
377         ao2_callback(sub->stasis_subs, OBJ_NODATA, get_message_count, &counter);
378
379         if (reason) {
380                 pj_cstr(&reason_str, reason);
381                 reason_str_ptr = &reason_str;
382         }
383         ast_str_append(&body, 0, "Messages-Waiting: %s\r\n", counter.new_msgs ? "yes" : "no");
384         ast_str_append(&body, 0, "Voice-Message: %d/%d (0/0)\r\n", counter.new_msgs, counter.old_msgs);
385         pj_cstr(&pj_body, ast_str_buffer(body));
386
387         ast_debug(5, "Sending  %s MWI NOTIFY to endpoint %s, new messages: %d, old messages: %d\n",
388                         sub->is_solicited ? "solicited" : "unsolicited", sub->id, counter.new_msgs,
389                         counter.old_msgs);
390
391         if (sub->is_solicited) {
392                 if (pjsip_mwi_notify(ast_sip_subscription_get_evsub(sub->sip_sub),
393                                 state,
394                                 NULL,
395                                 reason_str_ptr,
396                                 &mwi_type,
397                                 &pj_body,
398                                 &tdata) != PJ_SUCCESS) {
399                         ast_log(LOG_WARNING, "Unable to create MWI NOTIFY request to %s.\n", sub->id);
400                         return;
401                 }
402                 if (ast_sip_subscription_send_request(sub->sip_sub, tdata) != PJ_SUCCESS) {
403                         ast_log(LOG_WARNING, "Unable to send MWI NOTIFY request to %s\n", sub->id);
404                         return;
405                 }
406         } else {
407                 send_unsolicited_mwi_notify(sub, state, reason, &mwi_type, &pj_body);
408         }
409 }
410
411 static int unsubscribe_stasis(void *obj, void *arg, int flags)
412 {
413         struct mwi_stasis_subscription *mwi_stasis = obj;
414         if (mwi_stasis->stasis_sub) {
415                 ast_debug(3, "Removing stasis subscription to mailbox %s\n", mwi_stasis->mailbox);
416                 mwi_stasis->stasis_sub = stasis_unsubscribe(mwi_stasis->stasis_sub);
417         }
418         return CMP_MATCH;
419 }
420
421 static void mwi_subscription_shutdown(struct ast_sip_subscription *sub)
422 {
423         struct mwi_subscription *mwi_sub;
424         RAII_VAR(struct ast_datastore *, mwi_datastore,
425                         ast_sip_subscription_get_datastore(sub, "MWI datastore"), ao2_cleanup);
426
427         if (!mwi_datastore) {
428                 return;
429         }
430
431         mwi_sub = mwi_datastore->data;
432         ao2_callback(mwi_sub->stasis_subs, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe_stasis, NULL);
433 }
434
435 static struct ast_datastore_info mwi_ds_info = { };
436
437 static int add_mwi_datastore(struct mwi_subscription *sub)
438 {
439         RAII_VAR(struct ast_datastore *, mwi_datastore, NULL, ao2_cleanup);
440
441         mwi_datastore = ast_sip_subscription_alloc_datastore(&mwi_ds_info, "MWI datastore");
442         if (!mwi_datastore) {
443                 return -1;
444         }
445         mwi_datastore->data = sub;
446
447         ast_sip_subscription_add_datastore(sub->sip_sub, mwi_datastore);
448         return 0;
449 }
450
451 static struct ast_sip_subscription *mwi_new_subscribe(struct ast_sip_endpoint *endpoint,
452                 pjsip_rx_data *rdata)
453 {
454         RAII_VAR(struct ast_sip_aor *, aor, NULL, ao2_cleanup);
455         /* It's not obvious here, but the reference(s) to this subscription,
456          * once this function exits, is held by the stasis subscription(s)
457          * created in mwi_stasis_subscription_alloc()
458          */
459         RAII_VAR(struct mwi_subscription *, sub, NULL, ao2_cleanup);
460         pjsip_uri *ruri = rdata->msg_info.msg->line.req.uri;
461         pjsip_sip_uri *sip_ruri;
462         pjsip_evsub *evsub;
463         char aor_name[80];
464         char *mailboxes;
465         char *mailbox;
466
467         if (!PJSIP_URI_SCHEME_IS_SIP(ruri) && !PJSIP_URI_SCHEME_IS_SIPS(ruri)) {
468                 ast_log(LOG_WARNING, "Attempt to SUBSCRIBE to a non-SIP URI\n");
469                 return NULL;
470         }
471         sip_ruri = pjsip_uri_get_uri(ruri);
472         ast_copy_pj_str(aor_name, &sip_ruri->user, sizeof(aor_name));
473
474         aor = ast_sip_location_retrieve_aor(aor_name);
475         if (!aor) {
476                 ast_log(LOG_WARNING, "Unable to locate aor %s. MWI subscription failed.\n", aor_name);
477                 return NULL;
478         }
479
480         if (ast_strlen_zero(aor->mailboxes)) {
481                 ast_log(LOG_WARNING, "AOR %s has no configured mailboxes. MWI subscription failed\n", aor_name);
482                 return NULL;
483         }
484
485         sub = mwi_subscription_alloc(endpoint, AST_SIP_NOTIFIER, 1, rdata);
486         if (!sub) {
487                 return NULL;
488         }
489
490         if (add_mwi_datastore(sub)) {
491                 ast_log(LOG_WARNING, "Unable to allocate datastore on MWI subscription from %s\n", sub->id);
492                 return NULL;
493         }
494
495         mailboxes = ast_strdupa(aor->mailboxes);
496         while ((mailbox = strsep(&mailboxes, ","))) {
497                 RAII_VAR(struct mwi_stasis_subscription *, mwi_stasis_sub,
498                                 mwi_stasis_subscription_alloc(mailbox, sub), ao2_cleanup);
499                 if (mwi_stasis_sub) {
500                         ao2_link(sub->stasis_subs, mwi_stasis_sub);
501                 }
502         }
503
504         evsub = ast_sip_subscription_get_evsub(sub->sip_sub);
505         pjsip_evsub_accept(evsub, rdata, 200, NULL);
506         send_mwi_notify(sub, PJSIP_EVSUB_STATE_ACTIVE, NULL);
507
508         return sub->sip_sub;
509 }
510
511 static void mwi_resubscribe(struct ast_sip_subscription *sub,
512                 pjsip_rx_data *rdata, struct ast_sip_subscription_response_data *response_data)
513 {
514         pjsip_tx_data *tdata;
515
516         pjsip_mwi_current_notify(ast_sip_subscription_get_evsub(sub), &tdata);
517         ast_sip_subscription_send_request(sub, tdata);
518 }
519
520 static void mwi_subscription_timeout(struct ast_sip_subscription *sub)
521 {
522         struct mwi_subscription *mwi_sub;
523         RAII_VAR(struct ast_datastore *, mwi_datastore,
524                         ast_sip_subscription_get_datastore(sub, "MWI datastore"), ao2_cleanup);
525
526         if (!mwi_datastore) {
527                 return;
528         }
529
530
531         mwi_sub = mwi_datastore->data;
532
533         ast_log(LOG_NOTICE, "MWI subscription for %s has timed out.\n", mwi_sub->id);
534
535         send_mwi_notify(mwi_sub, PJSIP_EVSUB_STATE_TERMINATED, "timeout");
536 }
537
538 static void mwi_subscription_terminated(struct ast_sip_subscription *sub, pjsip_rx_data *rdata)
539 {
540         struct mwi_subscription *mwi_sub;
541         RAII_VAR(struct ast_datastore *, mwi_datastore,
542                         ast_sip_subscription_get_datastore(sub, "MWI datastore"), ao2_cleanup);
543
544         if (!mwi_datastore) {
545                 return;
546         }
547
548         mwi_sub = mwi_datastore->data;
549
550         ast_log(LOG_NOTICE, "MWI subscription for %s has been terminated\n", mwi_sub->id);
551
552         send_mwi_notify(mwi_sub, PJSIP_EVSUB_STATE_TERMINATED, NULL);
553 }
554
555 static void mwi_notify_response(struct ast_sip_subscription *sub, pjsip_rx_data *rdata)
556 {
557         /* We don't really care about NOTIFY responses for the moment */
558 }
559
560 static void mwi_notify_request(struct ast_sip_subscription *sub, pjsip_rx_data *rdata,
561                 struct ast_sip_subscription_response_data *response_data)
562 {
563         ast_log(LOG_WARNING, "Received an MWI NOTIFY request? This should not happen\n");
564 }
565
566 static int mwi_refresh_subscription(struct ast_sip_subscription *sub)
567 {
568         ast_log(LOG_WARNING, "Being told to refresh an MWI subscription? This should not happen\n");
569         return 0;
570 }
571
572 static int serialized_notify(void *userdata)
573 {
574         struct mwi_subscription *mwi_sub = userdata;
575
576         send_mwi_notify(mwi_sub, PJSIP_EVSUB_STATE_ACTIVE, NULL);
577         ao2_ref(mwi_sub, -1);
578         return 0;
579 }
580
581 static int serialized_cleanup(void *userdata)
582 {
583         struct mwi_subscription *mwi_sub = userdata;
584
585         /* This is getting rid of the reference that was added
586          * just before this serialized task was pushed.
587          */
588         ao2_cleanup(mwi_sub);
589         /* This is getting rid of the reference held by the
590          * stasis subscription
591          */
592         ao2_cleanup(mwi_sub);
593         return 0;
594 }
595
596 static void mwi_stasis_cb(void *userdata, struct stasis_subscription *sub,
597                 struct stasis_topic *topic, struct stasis_message *msg)
598 {
599         struct mwi_subscription *mwi_sub = userdata;
600
601         if (stasis_subscription_final_message(sub, msg)) {
602                 ao2_ref(mwi_sub, +1);
603                 ast_sip_push_task(NULL, serialized_cleanup, mwi_sub);
604                 return;
605         }
606
607         if (ast_mwi_state_type() == stasis_message_type(msg)) {
608                 struct ast_taskprocessor *serializer = mwi_sub->is_solicited ? ast_sip_subscription_get_serializer(mwi_sub->sip_sub) : NULL;
609                 ao2_ref(mwi_sub, +1);
610                 ast_sip_push_task(serializer, serialized_notify, mwi_sub);
611         }
612 }
613
614 static int create_mwi_subscriptions_for_endpoint(void *obj, void *arg, int flags)
615 {
616         RAII_VAR(struct mwi_subscription *, aggregate_sub, NULL, ao2_cleanup);
617         struct ast_sip_endpoint *endpoint = obj;
618         struct ao2_container *mwi_subscriptions = arg;
619         char *mailboxes;
620         char *mailbox;
621
622         if (ast_strlen_zero(endpoint->mailboxes)) {
623                 return 0;
624         }
625
626         if (endpoint->aggregate_mwi) {
627                 aggregate_sub = mwi_subscription_alloc(endpoint, AST_SIP_NOTIFIER, 0, NULL);
628                 if (!aggregate_sub) {
629                         return 0;
630                 }
631         }
632
633         mailboxes = ast_strdupa(endpoint->mailboxes);
634         while ((mailbox = strsep(&mailboxes, ","))) {
635                 struct mwi_subscription *sub = aggregate_sub ?:
636                         mwi_subscription_alloc(endpoint, AST_SIP_SUBSCRIBER, 0, NULL);
637                 RAII_VAR(struct mwi_stasis_subscription *, mwi_stasis_sub,
638                                 mwi_stasis_subscription_alloc(mailbox, sub), ao2_cleanup);
639                 if (mwi_stasis_sub) {
640                         ao2_link(sub->stasis_subs, mwi_stasis_sub);
641                 }
642                 if (!aggregate_sub) {
643                         ao2_link(mwi_subscriptions, sub);
644                         ao2_cleanup(sub);
645                 }
646         }
647         ao2_link(mwi_subscriptions, aggregate_sub);
648         return 0;
649 }
650
651 static int unsubscribe(void *obj, void *arg, int flags)
652 {
653         struct mwi_subscription *mwi_sub = obj;
654
655         ao2_callback(mwi_sub->stasis_subs, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe_stasis, NULL);
656         return CMP_MATCH;
657 }
658
659 static void create_mwi_subscriptions(void)
660 {
661         struct ao2_container *mwi_subscriptions = ao2_container_alloc(MWI_BUCKETS, mwi_sub_hash, mwi_sub_cmp);
662         RAII_VAR(struct ao2_container *, old_mwi_subscriptions, ao2_global_obj_ref(unsolicited_mwi), ao2_cleanup);
663         RAII_VAR(struct ao2_container *, endpoints, ast_sorcery_retrieve_by_fields(
664                                 ast_sip_get_sorcery(), "endpoint", AST_RETRIEVE_FLAG_MULTIPLE | AST_RETRIEVE_FLAG_ALL, NULL),
665                         ao2_cleanup);
666
667         if (!mwi_subscriptions) {
668                 return;
669         }
670
671         /* We remove all the old stasis subscriptions first before applying the new configuration. This
672          * prevents a situation where there might be multiple overlapping stasis subscriptions for an
673          * endpoint for mailboxes. Though there may be mailbox changes during the gap between unsubscribing
674          * and resubscribing, up-to-date mailbox state will be sent out to the endpoint when the
675          * new stasis subscription is established
676          */
677         if (old_mwi_subscriptions) {
678                 ao2_callback(old_mwi_subscriptions, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe, NULL);
679         }
680         ao2_callback(endpoints, OBJ_NODATA, create_mwi_subscriptions_for_endpoint, mwi_subscriptions);
681         ao2_global_obj_replace_unref(unsolicited_mwi, mwi_subscriptions);
682 }
683
684 static int reload(void)
685 {
686         create_mwi_subscriptions();
687         return 0;
688 }
689
690 static int load_module(void)
691 {
692         if (ast_sip_register_subscription_handler(&mwi_handler)) {
693                 return AST_MODULE_LOAD_DECLINE;
694         }
695         create_mwi_subscriptions();
696         return AST_MODULE_LOAD_SUCCESS;
697 }
698
699 static int unload_module(void)
700 {
701         RAII_VAR(struct ao2_container *, mwi_subscriptions, ao2_global_obj_ref(unsolicited_mwi), ao2_cleanup);
702         if (mwi_subscriptions) {
703                 ao2_callback(mwi_subscriptions, OBJ_UNLINK | OBJ_NODATA | OBJ_MULTIPLE, unsubscribe, NULL);
704                 ao2_global_obj_release(unsolicited_mwi);
705         }
706         ast_sip_unregister_subscription_handler(&mwi_handler);
707         return 0;
708 }
709
710 AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_LOAD_ORDER, "SIP MWI resource",
711                 .load = load_module,
712                 .unload = unload_module,
713                 .reload = reload,
714                 .load_pri = AST_MODPRI_CHANNEL_DEPEND,
715 );