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