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