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