ARI: Add ability to raise arbitrary User Events
[asterisk/asterisk.git] / main / stasis_cache.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 Message API.
22  *
23  * \author David M. Lee, II <dlee@digium.com>
24  */
25
26 /*** MODULEINFO
27         <support_level>core</support_level>
28  ***/
29
30 #include "asterisk.h"
31
32 ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
33
34 #include "asterisk/astobj2.h"
35 #include "asterisk/hashtab.h"
36 #include "asterisk/stasis_internal.h"
37 #include "asterisk/stasis.h"
38 #include "asterisk/utils.h"
39 #include "asterisk/vector.h"
40
41 #ifdef LOW_MEMORY
42 #define NUM_CACHE_BUCKETS 17
43 #else
44 #define NUM_CACHE_BUCKETS 563
45 #endif
46
47 /*! \internal */
48 struct stasis_cache {
49         struct ao2_container *entries;
50         snapshot_get_id id_fn;
51         cache_aggregate_calc_fn aggregate_calc_fn;
52         cache_aggregate_publish_fn aggregate_publish_fn;
53 };
54
55 /*! \internal */
56 struct stasis_caching_topic {
57         struct stasis_cache *cache;
58         struct stasis_topic *topic;
59         struct stasis_topic *original_topic;
60         struct stasis_subscription *sub;
61 };
62
63 static void stasis_caching_topic_dtor(void *obj)
64 {
65         struct stasis_caching_topic *caching_topic = obj;
66
67         /* Caching topics contain subscriptions, and must be manually
68          * unsubscribed. */
69         ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
70         /* If there are any messages in flight to this subscription; that would
71          * be bad. */
72         ast_assert(stasis_subscription_is_done(caching_topic->sub));
73
74         ao2_cleanup(caching_topic->sub);
75         caching_topic->sub = NULL;
76         ao2_cleanup(caching_topic->cache);
77         caching_topic->cache = NULL;
78         ao2_cleanup(caching_topic->topic);
79         caching_topic->topic = NULL;
80         ao2_cleanup(caching_topic->original_topic);
81         caching_topic->original_topic = NULL;
82 }
83
84 struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
85 {
86         return caching_topic->topic;
87 }
88
89 struct stasis_caching_topic *stasis_caching_unsubscribe(struct stasis_caching_topic *caching_topic)
90 {
91         if (!caching_topic) {
92                 return NULL;
93         }
94
95         /*
96          * The subscription may hold the last reference to this caching
97          * topic, but we want to make sure the unsubscribe finishes
98          * before kicking of the caching topic's dtor.
99          */
100         ao2_ref(caching_topic, +1);
101
102         if (stasis_subscription_is_subscribed(caching_topic->sub)) {
103                 /*
104                  * Increment the reference to hold on to it past the
105                  * unsubscribe. Will be cleaned up in dtor.
106                  */
107                 ao2_ref(caching_topic->sub, +1);
108                 stasis_unsubscribe(caching_topic->sub);
109         } else {
110                 ast_log(LOG_ERROR, "stasis_caching_topic unsubscribed multiple times\n");
111         }
112         ao2_cleanup(caching_topic);
113         return NULL;
114 }
115
116 struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_caching_topic *caching_topic)
117 {
118         if (!caching_topic) {
119                 return NULL;
120         }
121
122         /* Hold a ref past the unsubscribe */
123         ao2_ref(caching_topic, +1);
124         stasis_caching_unsubscribe(caching_topic);
125         stasis_subscription_join(caching_topic->sub);
126         ao2_cleanup(caching_topic);
127         return NULL;
128 }
129
130 struct cache_entry_key {
131         struct stasis_message_type *type;
132         const char *id;
133 };
134
135 struct stasis_cache_entry {
136         struct cache_entry_key key;
137         /*! Aggregate snapshot of the stasis cache. */
138         struct stasis_message *aggregate;
139         /*! Local entity snapshot of the stasis event. */
140         struct stasis_message *local;
141         /*! Remote entity snapshots of the stasis event. */
142         AST_VECTOR(, struct stasis_message *) remote;
143 };
144
145 static void cache_entry_dtor(void *obj)
146 {
147         struct stasis_cache_entry *entry = obj;
148         size_t idx;
149
150         ao2_cleanup(entry->key.type);
151         entry->key.type = NULL;
152         ast_free((char *) entry->key.id);
153         entry->key.id = NULL;
154
155         ao2_cleanup(entry->aggregate);
156         entry->aggregate = NULL;
157         ao2_cleanup(entry->local);
158         entry->local = NULL;
159
160         for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
161                 struct stasis_message *remote;
162
163                 remote = AST_VECTOR_GET(&entry->remote, idx);
164                 ao2_cleanup(remote);
165         }
166         AST_VECTOR_FREE(&entry->remote);
167 }
168
169 static struct stasis_cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
170 {
171         struct stasis_cache_entry *entry;
172         int is_remote;
173
174         ast_assert(type != NULL);
175         ast_assert(id != NULL);
176         ast_assert(snapshot != NULL);
177
178         entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
179                 AO2_ALLOC_OPT_LOCK_NOLOCK);
180         if (!entry) {
181                 return NULL;
182         }
183
184         entry->key.id = ast_strdup(id);
185         if (!entry->key.id) {
186                 ao2_cleanup(entry);
187                 return NULL;
188         }
189         entry->key.type = ao2_bump(type);
190
191         is_remote = ast_eid_cmp(&ast_eid_default, stasis_message_eid(snapshot)) ? 1 : 0;
192         if (AST_VECTOR_INIT(&entry->remote, is_remote)) {
193                 ao2_cleanup(entry);
194                 return NULL;
195         }
196
197         if (is_remote) {
198                 if (AST_VECTOR_APPEND(&entry->remote, snapshot)) {
199                         ao2_cleanup(entry);
200                         return NULL;
201                 }
202         } else {
203                 entry->local = snapshot;
204         }
205         ao2_bump(snapshot);
206
207         return entry;
208 }
209
210 static int cache_entry_hash(const void *obj, int flags)
211 {
212         const struct stasis_cache_entry *object;
213         const struct cache_entry_key *key;
214         int hash = 0;
215
216         switch (flags & OBJ_SEARCH_MASK) {
217         case OBJ_SEARCH_KEY:
218                 key = obj;
219                 break;
220         case OBJ_SEARCH_OBJECT:
221                 object = obj;
222                 key = &object->key;
223                 break;
224         default:
225                 /* Hash can only work on something with a full key. */
226                 ast_assert(0);
227                 return 0;
228         }
229
230         hash += ast_hashtab_hash_string(stasis_message_type_name(key->type));
231         hash += ast_hashtab_hash_string(key->id);
232         return hash;
233 }
234
235 static int cache_entry_cmp(void *obj, void *arg, int flags)
236 {
237         const struct stasis_cache_entry *object_left = obj;
238         const struct stasis_cache_entry *object_right = arg;
239         const struct cache_entry_key *right_key = arg;
240         int cmp;
241
242         switch (flags & OBJ_SEARCH_MASK) {
243         case OBJ_SEARCH_OBJECT:
244                 right_key = &object_right->key;
245                 /* Fall through */
246         case OBJ_SEARCH_KEY:
247                 cmp = object_left->key.type != right_key->type
248                         || strcmp(object_left->key.id, right_key->id);
249                 break;
250         case OBJ_SEARCH_PARTIAL_KEY:
251                 /* Not supported by container */
252                 ast_assert(0);
253                 cmp = -1;
254                 break;
255         default:
256                 /*
257                  * What arg points to is specific to this traversal callback
258                  * and has no special meaning to astobj2.
259                  */
260                 cmp = 0;
261                 break;
262         }
263         if (cmp) {
264                 return 0;
265         }
266         /*
267          * At this point the traversal callback is identical to a sorted
268          * container.
269          */
270         return CMP_MATCH;
271 }
272
273 static void cache_dtor(void *obj)
274 {
275         struct stasis_cache *cache = obj;
276
277         ao2_cleanup(cache->entries);
278         cache->entries = NULL;
279 }
280
281 struct stasis_cache *stasis_cache_create_full(snapshot_get_id id_fn,
282         cache_aggregate_calc_fn aggregate_calc_fn,
283         cache_aggregate_publish_fn aggregate_publish_fn)
284 {
285         struct stasis_cache *cache;
286
287         cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
288                 AO2_ALLOC_OPT_LOCK_NOLOCK);
289         if (!cache) {
290                 return NULL;
291         }
292
293         cache->entries = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
294                 NUM_CACHE_BUCKETS, cache_entry_hash, NULL, cache_entry_cmp);
295         if (!cache->entries) {
296                 ao2_cleanup(cache);
297                 return NULL;
298         }
299
300         cache->id_fn = id_fn;
301         cache->aggregate_calc_fn = aggregate_calc_fn;
302         cache->aggregate_publish_fn = aggregate_publish_fn;
303
304         return cache;
305 }
306
307 struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
308 {
309         return stasis_cache_create_full(id_fn, NULL, NULL);
310 }
311
312 struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry)
313 {
314         return entry->aggregate;
315 }
316
317 struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry)
318 {
319         return entry->local;
320 }
321
322 struct stasis_message *stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx)
323 {
324         if (idx < AST_VECTOR_SIZE(&entry->remote)) {
325                 return AST_VECTOR_GET(&entry->remote, idx);
326         }
327         return NULL;
328 }
329
330 /*!
331  * \internal
332  * \brief Find the cache entry in the cache entries container.
333  *
334  * \param entries Container of cached entries.
335  * \param type Type of message to retrieve the cache entry.
336  * \param id Identity of the snapshot to retrieve the cache entry.
337  *
338  * \note The entries container is already locked.
339  *
340  * \retval Cache-entry on success.
341  * \retval NULL Not in cache.
342  */
343 static struct stasis_cache_entry *cache_find(struct ao2_container *entries, struct stasis_message_type *type, const char *id)
344 {
345         struct cache_entry_key search_key;
346         struct stasis_cache_entry *entry;
347
348         search_key.type = type;
349         search_key.id = id;
350         entry = ao2_find(entries, &search_key, OBJ_SEARCH_KEY | OBJ_NOLOCK);
351
352         /* Ensure that what we looked for is what we found. */
353         ast_assert(!entry
354                 || (!strcmp(stasis_message_type_name(entry->key.type),
355                         stasis_message_type_name(type)) && !strcmp(entry->key.id, id)));
356         return entry;
357 }
358
359 /*!
360  * \internal
361  * \brief Remove the stasis snapshot in the cache entry determined by eid.
362  *
363  * \param entries Container of cached entries.
364  * \param cached_entry The entry to remove the snapshot from.
365  * \param eid Which snapshot in the cached entry.
366  *
367  * \note The entries container is already locked.
368  *
369  * \return Previous stasis entry snapshot.
370  */
371 static struct stasis_message *cache_remove(struct ao2_container *entries, struct stasis_cache_entry *cached_entry, const struct ast_eid *eid)
372 {
373         struct stasis_message *old_snapshot;
374         int is_remote;
375
376         is_remote = ast_eid_cmp(eid, &ast_eid_default);
377         if (!is_remote) {
378                 old_snapshot = cached_entry->local;
379                 cached_entry->local = NULL;
380         } else {
381                 int idx;
382
383                 old_snapshot = NULL;
384                 for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
385                         struct stasis_message *cur;
386
387                         cur = AST_VECTOR_GET(&cached_entry->remote, idx);
388                         if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
389                                 old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
390                                 break;
391                         }
392                 }
393         }
394
395         if (!cached_entry->local && !AST_VECTOR_SIZE(&cached_entry->remote)) {
396                 ao2_unlink_flags(entries, cached_entry, OBJ_NOLOCK);
397         }
398
399         return old_snapshot;
400 }
401
402 /*!
403  * \internal
404  * \brief Update the stasis snapshot in the cache entry determined by eid.
405  *
406  * \param cached_entry The entry to remove the snapshot from.
407  * \param eid Which snapshot in the cached entry.
408  * \param new_snapshot Snapshot to replace the old snapshot.
409  *
410  * \return Previous stasis entry snapshot.
411  */
412 static struct stasis_message *cache_udpate(struct stasis_cache_entry *cached_entry, const struct ast_eid *eid, struct stasis_message *new_snapshot)
413 {
414         struct stasis_message *old_snapshot;
415         int is_remote;
416         int idx;
417
418         is_remote = ast_eid_cmp(eid, &ast_eid_default);
419         if (!is_remote) {
420                 old_snapshot = cached_entry->local;
421                 cached_entry->local = ao2_bump(new_snapshot);
422                 return old_snapshot;
423         }
424
425         old_snapshot = NULL;
426         for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
427                 struct stasis_message *cur;
428
429                 cur = AST_VECTOR_GET(&cached_entry->remote, idx);
430                 if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
431                         old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
432                         break;
433                 }
434         }
435         if (!AST_VECTOR_APPEND(&cached_entry->remote, new_snapshot)) {
436                 ao2_bump(new_snapshot);
437         }
438
439         return old_snapshot;
440 }
441
442 struct cache_put_snapshots {
443         /*! Old cache eid snapshot. */
444         struct stasis_message *old;
445         /*! Old cache aggregate snapshot. */
446         struct stasis_message *aggregate_old;
447         /*! New cache aggregate snapshot. */
448         struct stasis_message *aggregate_new;
449 };
450
451 static struct cache_put_snapshots cache_put(struct stasis_cache *cache,
452         struct stasis_message_type *type, const char *id, const struct ast_eid *eid,
453         struct stasis_message *new_snapshot)
454 {
455         struct stasis_cache_entry *cached_entry;
456         struct cache_put_snapshots snapshots;
457
458         ast_assert(cache->entries != NULL);
459         ast_assert(eid != NULL);/* Aggregate snapshots not allowed to be put directly. */
460         ast_assert(new_snapshot == NULL ||
461                 type == stasis_message_type(new_snapshot));
462
463         memset(&snapshots, 0, sizeof(snapshots));
464
465         ao2_wrlock(cache->entries);
466
467         cached_entry = cache_find(cache->entries, type, id);
468
469         /* Update the eid snapshot. */
470         if (!new_snapshot) {
471                 /* Remove snapshot from cache */
472                 if (cached_entry) {
473                         snapshots.old = cache_remove(cache->entries, cached_entry, eid);
474                 }
475         } else if (cached_entry) {
476                 /* Update snapshot in cache */
477                 snapshots.old = cache_udpate(cached_entry, eid, new_snapshot);
478         } else {
479                 /* Insert into the cache */
480                 cached_entry = cache_entry_create(type, id, new_snapshot);
481                 if (cached_entry) {
482                         ao2_link_flags(cache->entries, cached_entry, OBJ_NOLOCK);
483                 }
484         }
485
486         /* Update the aggregate snapshot. */
487         if (cache->aggregate_calc_fn && cached_entry) {
488                 snapshots.aggregate_new = cache->aggregate_calc_fn(cached_entry, new_snapshot);
489                 snapshots.aggregate_old = cached_entry->aggregate;
490                 cached_entry->aggregate = ao2_bump(snapshots.aggregate_new);
491         }
492
493         ao2_unlock(cache->entries);
494
495         ao2_cleanup(cached_entry);
496         return snapshots;
497 }
498
499 /*!
500  * \internal
501  * \brief Dump all entity snapshots in the cache entry into the given container.
502  *
503  * \param snapshots Container to put all snapshots in the cache entry.
504  * \param entry Cache entry to use.
505  *
506  * \retval 0 on success.
507  * \retval non-zero on error.
508  */
509 static int cache_entry_dump(struct ao2_container *snapshots, const struct stasis_cache_entry *entry)
510 {
511         int idx;
512         int err = 0;
513
514         ast_assert(snapshots != NULL);
515         ast_assert(entry != NULL);
516
517         /* The aggregate snapshot is not a snapshot from an entity. */
518
519         if (entry->local) {
520                 err |= !ao2_link(snapshots, entry->local);
521         }
522
523         for (idx = 0; !err && idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
524                 struct stasis_message *snapshot;
525
526                 snapshot = AST_VECTOR_GET(&entry->remote, idx);
527                 err |= !ao2_link(snapshots, snapshot);
528         }
529
530         return err;
531 }
532
533 struct ao2_container *stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
534 {
535         struct stasis_cache_entry *cached_entry;
536         struct ao2_container *found;
537
538         ast_assert(cache != NULL);
539         ast_assert(cache->entries != NULL);
540         ast_assert(type != NULL);
541         ast_assert(id != NULL);
542
543         found = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
544         if (!found) {
545                 return NULL;
546         }
547
548         ao2_rdlock(cache->entries);
549
550         cached_entry = cache_find(cache->entries, type, id);
551         if (cached_entry && cache_entry_dump(found, cached_entry)) {
552                 ao2_cleanup(found);
553                 found = NULL;
554         }
555
556         ao2_unlock(cache->entries);
557
558         ao2_cleanup(cached_entry);
559         return found;
560 }
561
562 /*!
563  * \internal
564  * \brief Retrieve an item from the cache entry for a specific eid.
565  *
566  * \param entry Cache entry to use.
567  * \param eid Specific entity id to retrieve.  NULL for aggregate.
568  *
569  * \note The returned snapshot has not had its reference bumped.
570  *
571  * \retval Snapshot from the cache.
572  * \retval \c NULL if snapshot is not found.
573  */
574 static struct stasis_message *cache_entry_by_eid(const struct stasis_cache_entry *entry, const struct ast_eid *eid)
575 {
576         int is_remote;
577         int idx;
578
579         if (!eid) {
580                 /* Get aggregate. */
581                 return entry->aggregate;
582         }
583
584         /* Get snapshot with specific eid. */
585         is_remote = ast_eid_cmp(eid, &ast_eid_default);
586         if (!is_remote) {
587                 return entry->local;
588         }
589
590         for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
591                 struct stasis_message *cur;
592
593                 cur = AST_VECTOR_GET(&entry->remote, idx);
594                 if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
595                         return cur;
596                 }
597         }
598
599         return NULL;
600 }
601
602 struct stasis_message *stasis_cache_get_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const char *id, const struct ast_eid *eid)
603 {
604         struct stasis_cache_entry *cached_entry;
605         struct stasis_message *snapshot = NULL;
606
607         ast_assert(cache != NULL);
608         ast_assert(cache->entries != NULL);
609         ast_assert(type != NULL);
610         ast_assert(id != NULL);
611
612         ao2_rdlock(cache->entries);
613
614         cached_entry = cache_find(cache->entries, type, id);
615         if (cached_entry) {
616                 snapshot = cache_entry_by_eid(cached_entry, eid);
617                 ao2_bump(snapshot);
618         }
619
620         ao2_unlock(cache->entries);
621
622         ao2_cleanup(cached_entry);
623         return snapshot;
624 }
625
626 struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
627 {
628         return stasis_cache_get_by_eid(cache, type, id, &ast_eid_default);
629 }
630
631 struct cache_dump_data {
632         struct ao2_container *container;
633         struct stasis_message_type *type;
634         const struct ast_eid *eid;
635 };
636
637 static int cache_dump_by_eid_cb(void *obj, void *arg, int flags)
638 {
639         struct cache_dump_data *cache_dump = arg;
640         struct stasis_cache_entry *entry = obj;
641
642         if (!cache_dump->type || entry->key.type == cache_dump->type) {
643                 struct stasis_message *snapshot;
644
645                 snapshot = cache_entry_by_eid(entry, cache_dump->eid);
646                 if (snapshot) {
647                         if (!ao2_link(cache_dump->container, snapshot)) {
648                                 ao2_cleanup(cache_dump->container);
649                                 cache_dump->container = NULL;
650                                 return CMP_STOP;
651                         }
652                 }
653         }
654
655         return 0;
656 }
657
658 struct ao2_container *stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
659 {
660         struct cache_dump_data cache_dump;
661
662         ast_assert(cache != NULL);
663         ast_assert(cache->entries != NULL);
664
665         cache_dump.eid = eid;
666         cache_dump.type = type;
667         cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
668         if (!cache_dump.container) {
669                 return NULL;
670         }
671
672         ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_by_eid_cb, &cache_dump);
673         return cache_dump.container;
674 }
675
676 struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
677 {
678         return stasis_cache_dump_by_eid(cache, type, &ast_eid_default);
679 }
680
681 static int cache_dump_all_cb(void *obj, void *arg, int flags)
682 {
683         struct cache_dump_data *cache_dump = arg;
684         struct stasis_cache_entry *entry = obj;
685
686         if (!cache_dump->type || entry->key.type == cache_dump->type) {
687                 if (cache_entry_dump(cache_dump->container, entry)) {
688                         ao2_cleanup(cache_dump->container);
689                         cache_dump->container = NULL;
690                         return CMP_STOP;
691                 }
692         }
693
694         return 0;
695 }
696
697 struct ao2_container *stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type)
698 {
699         struct cache_dump_data cache_dump;
700
701         ast_assert(cache != NULL);
702         ast_assert(cache->entries != NULL);
703
704         cache_dump.eid = NULL;
705         cache_dump.type = type;
706         cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
707         if (!cache_dump.container) {
708                 return NULL;
709         }
710
711         ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_all_cb, &cache_dump);
712         return cache_dump.container;
713 }
714
715 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
716 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type);
717
718 struct stasis_message *stasis_cache_clear_create(struct stasis_message *id_message)
719 {
720         return stasis_message_create(stasis_cache_clear_type(), id_message);
721 }
722
723 static void stasis_cache_update_dtor(void *obj)
724 {
725         struct stasis_cache_update *update = obj;
726
727         ao2_cleanup(update->old_snapshot);
728         update->old_snapshot = NULL;
729         ao2_cleanup(update->new_snapshot);
730         update->new_snapshot = NULL;
731         ao2_cleanup(update->type);
732         update->type = NULL;
733 }
734
735 static struct stasis_message *update_create(struct stasis_message *old_snapshot, struct stasis_message *new_snapshot)
736 {
737         struct stasis_cache_update *update;
738         struct stasis_message *msg;
739
740         ast_assert(old_snapshot != NULL || new_snapshot != NULL);
741
742         update = ao2_alloc_options(sizeof(*update), stasis_cache_update_dtor,
743                 AO2_ALLOC_OPT_LOCK_NOLOCK);
744         if (!update) {
745                 return NULL;
746         }
747
748         if (old_snapshot) {
749                 ao2_ref(old_snapshot, +1);
750                 update->old_snapshot = old_snapshot;
751                 if (!new_snapshot) {
752                         ao2_ref(stasis_message_type(old_snapshot), +1);
753                         update->type = stasis_message_type(old_snapshot);
754                 }
755         }
756         if (new_snapshot) {
757                 ao2_ref(new_snapshot, +1);
758                 update->new_snapshot = new_snapshot;
759                 ao2_ref(stasis_message_type(new_snapshot), +1);
760                 update->type = stasis_message_type(new_snapshot);
761         }
762
763         msg = stasis_message_create(stasis_cache_update_type(), update);
764
765         ao2_cleanup(update);
766         return msg;
767 }
768
769 static void caching_topic_exec(void *data, struct stasis_subscription *sub,
770         struct stasis_message *message)
771 {
772         struct stasis_caching_topic *caching_topic_needs_unref;
773         struct stasis_caching_topic *caching_topic = data;
774         struct stasis_message *msg;
775         struct stasis_message *msg_put;
776         struct stasis_message_type *msg_type;
777         const struct ast_eid *msg_eid;
778         const char *msg_id;
779
780         ast_assert(caching_topic != NULL);
781         ast_assert(caching_topic->topic != NULL);
782         ast_assert(caching_topic->cache != NULL);
783         ast_assert(caching_topic->cache->id_fn != NULL);
784
785         if (stasis_subscription_final_message(sub, message)) {
786                 caching_topic_needs_unref = caching_topic;
787         } else {
788                 caching_topic_needs_unref = NULL;
789         }
790
791         msg_type = stasis_message_type(message);
792         if (stasis_cache_clear_type() == msg_type) {
793                 /* Cache clear event. */
794                 msg_put = NULL;
795                 msg = stasis_message_data(message);
796                 msg_type = stasis_message_type(msg);
797         } else {
798                 /* Normal cache update event. */
799                 msg_put = message;
800                 msg = message;
801         }
802         ast_assert(msg_type != NULL);
803
804         msg_eid = stasis_message_eid(msg);/* msg_eid is NULL for aggregate message. */
805         msg_id = caching_topic->cache->id_fn(msg);
806         if (msg_id && msg_eid) {
807                 struct stasis_message *update;
808                 struct cache_put_snapshots snapshots;
809
810                 /* Update the cache */
811                 snapshots = cache_put(caching_topic->cache, msg_type, msg_id, msg_eid, msg_put);
812                 if (snapshots.old || msg_put) {
813                         update = update_create(snapshots.old, msg_put);
814                         if (update) {
815                                 stasis_publish(caching_topic->topic, update);
816                         }
817                         ao2_cleanup(update);
818                 } else {
819                         ast_log(LOG_ERROR,
820                                 "Attempting to remove an item from the %s cache that isn't there: %s %s\n",
821                                 stasis_topic_name(caching_topic->topic),
822                                 stasis_message_type_name(msg_type), msg_id);
823                 }
824
825                 if (snapshots.aggregate_old != snapshots.aggregate_new) {
826                         if (snapshots.aggregate_new && caching_topic->cache->aggregate_publish_fn) {
827                                 caching_topic->cache->aggregate_publish_fn(caching_topic->original_topic,
828                                         snapshots.aggregate_new);
829                         }
830                         update = update_create(snapshots.aggregate_old, snapshots.aggregate_new);
831                         if (update) {
832                                 stasis_publish(caching_topic->topic, update);
833                         }
834                         ao2_cleanup(update);
835                 }
836
837                 ao2_cleanup(snapshots.old);
838                 ao2_cleanup(snapshots.aggregate_old);
839                 ao2_cleanup(snapshots.aggregate_new);
840         }
841
842         ao2_cleanup(caching_topic_needs_unref);
843 }
844
845 struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
846 {
847         RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, ao2_cleanup);
848         struct stasis_subscription *sub;
849         RAII_VAR(char *, new_name, NULL, ast_free);
850         int ret;
851
852         ret = ast_asprintf(&new_name, "%s-cached", stasis_topic_name(original_topic));
853         if (ret < 0) {
854                 return NULL;
855         }
856
857         caching_topic = ao2_alloc_options(sizeof(*caching_topic),
858                 stasis_caching_topic_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
859         if (caching_topic == NULL) {
860                 return NULL;
861         }
862
863         caching_topic->topic = stasis_topic_create(new_name);
864         if (caching_topic->topic == NULL) {
865                 return NULL;
866         }
867
868         ao2_ref(cache, +1);
869         caching_topic->cache = cache;
870
871         sub = internal_stasis_subscribe(original_topic, caching_topic_exec, caching_topic, 0);
872         if (sub == NULL) {
873                 return NULL;
874         }
875
876         ao2_ref(original_topic, +1);
877         caching_topic->original_topic = original_topic;
878
879         /* This is for the reference contained in the subscription above */
880         ao2_ref(caching_topic, +1);
881         caching_topic->sub = sub;
882
883         /* The subscription holds the reference, so no additional ref bump. */
884         return caching_topic;
885 }
886
887 static void stasis_cache_cleanup(void)
888 {
889         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
890         STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
891 }
892
893 int stasis_cache_init(void)
894 {
895         ast_register_cleanup(stasis_cache_cleanup);
896
897         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_clear_type) != 0) {
898                 return -1;
899         }
900
901         if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_update_type) != 0) {
902                 return -1;
903         }
904
905         return 0;
906 }
907