stasis cache: Enhance to keep track of an item from different entities.

A stasis cache entry now contains more than a single message/snapshot.  It
contains messages/snapshots for the local entity as well as any remote
entities that post to the cached item.  In addition callbacks can be
supplied when the cache is created to compute and post the aggregate
message/snapshot representing all entities stored in the cache entry.

* All stasis messages now have an eid to indicate what entity posted it.

* The stasis cache enhancements allow device state to cache and aggregate
the device states from local and remote entities in a single operation.
The cached aggregate device state is available immediately after it is
posted to the stasis bus.  This improves performance by eliminating a
cache dump and associated ao2 container traversals to calculate the
aggregate state.

(closes issue ASTERISK-23204)
Reported by: Mark Michelson

Review: https://reviewboard.asterisk.org/r/3281/
........

Merged revisions 410184 from http://svn.asterisk.org/svn/asterisk/branches/12


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@410185 65c4cc65-6c06-0410-ace0-fbb531ad65f3

4 files changed, 744 insertions, 278 deletions

diff --git a/main/app.c b/main/app.c
index 6e1a176c6..ce0d16cdd 100644
--- a/main/app.c
+++ b/main/app.c
@@ -2821,7 +2821,6 @@ struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context)
 	return mwi_state;
 }
 
-
 int ast_publish_mwi_state_full(
 			const char *mailbox,
 			const char *context,
@@ -2857,10 +2856,19 @@ int ast_publish_mwi_state_full(
 	if (eid) {
 		mwi_state->eid = *eid;
 	} else {
-		ast_set_default_eid(&mwi_state->eid);
+		mwi_state->eid = ast_eid_default;
 	}
 
-	message = stasis_message_create(ast_mwi_state_type(), mwi_state);
+	/*
+	 * As far as stasis is concerned, all MWI events are internal.
+	 *
+	 * We may in the future want to make MWI aggregate internal/external
+	 * message counts similar to how device state aggregates state.
+	 */
+	message = stasis_message_create_full(ast_mwi_state_type(), mwi_state, &ast_eid_default);
+	if (!message) {
+		return -1;
+	}
 
 	mailbox_specific_topic = ast_mwi_topic(mwi_state->uniqueid);
 	if (!mailbox_specific_topic) {
@@ -2911,6 +2919,7 @@ struct stasis_message *ast_mwi_blob_create(struct ast_mwi_state *mwi_state,
 	ao2_ref(obj->mwi_state, +1);
 	obj->blob = ast_json_ref(blob);
 
+	/* This is not a normal MWI event.  Only used by the MinivmNotify app. */
 	msg = stasis_message_create(message_type, obj);
 	if (!msg) {
 		return NULL;
diff --git a/main/devicestate.c b/main/devicestate.c
index 1f0c968aa..5c2340863 100644
--- a/main/devicestate.c
+++ b/main/devicestate.c
@@ -282,16 +282,20 @@ enum ast_device_state ast_parse_device_state(const char *device)
 
 static enum ast_device_state devstate_cached(const char *device)
 {
-	RAII_VAR(struct stasis_message *, cached_msg, NULL, ao2_cleanup);
+	struct stasis_message *cached_msg;
 	struct ast_device_state_message *device_state;
+	enum ast_device_state state;
 
-	cached_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device);
+	cached_msg = stasis_cache_get_by_eid(ast_device_state_cache(),
+		ast_device_state_message_type(), device, NULL);
 	if (!cached_msg) {
 		return AST_DEVICE_UNKNOWN;
 	}
 	device_state = stasis_message_data(cached_msg);
+	state = device_state->state;
+	ao2_cleanup(cached_msg);
 
-	return device_state->state;
+	return state;
 }
 
 /*! \brief Check device state through channel specific function or generic function */
@@ -522,148 +526,62 @@ static void *do_devstate_changes(void *data)
 	return NULL;
 }
 
-#define MAX_SERVERS 64
-static int devstate_change_aggregator_cb(void *obj, void *arg, void *data, int flags)
-{
-	struct stasis_message *msg = obj;
-	struct ast_devstate_aggregate *aggregate = arg;
-	char *device = data;
-	struct ast_device_state_message *device_state = stasis_message_data(msg);
-
-	if (!device_state->eid || strcmp(device, device_state->device)) {
-		/* ignore aggregate states and devices that don't match */
-		return 0;
-	}
-	ast_debug(1, "Adding per-server state of '%s' for '%s'\n",
-		ast_devstate2str(device_state->state), device);
-	ast_devstate_aggregate_add(aggregate, device_state->state);
-	return 0;
-}
-
-static void device_state_dtor(void *obj)
-{
-	struct ast_device_state_message *device_state = obj;
-	ast_string_field_free_memory(device_state);
-	ast_free(device_state->eid);
-}
-
 static struct ast_device_state_message *device_state_alloc(const char *device, enum ast_device_state state, enum ast_devstate_cache cachable, const struct ast_eid *eid)
 {
-	RAII_VAR(struct ast_device_state_message *, new_device_state, ao2_alloc(sizeof(*new_device_state), device_state_dtor), ao2_cleanup);
+	struct ast_device_state_message *new_device_state;
+	char *pos;
+	size_t stuff_len;
+
+	ast_assert(!ast_strlen_zero(device));
 
-	if (!new_device_state || ast_string_field_init(new_device_state, 256)) {
+	stuff_len = strlen(device) + 1;
+	if (eid) {
+		stuff_len += sizeof(*eid);
+	}
+	new_device_state = ao2_alloc_options(sizeof(*new_device_state) + stuff_len, NULL,
+		AO2_ALLOC_OPT_LOCK_NOLOCK);
+	if (!new_device_state) {
 		return NULL;
 	}
 
-	ast_string_field_set(new_device_state, device, device);
-	new_device_state->state = state;
-	new_device_state->cachable = cachable;
-
 	if (eid) {
-		char eid_str[20];
-		struct ast_str *cache_id = ast_str_alloca(256);
-
-		new_device_state->eid = ast_malloc(sizeof(*eid));
-		if (!new_device_state->eid) {
-			return NULL;
-		}
-
-		*new_device_state->eid = *eid;
-		ast_eid_to_str(eid_str, sizeof(eid_str), new_device_state->eid);
-		ast_str_set(&cache_id, 0, "%s%s", eid_str, device);
-		ast_string_field_set(new_device_state, cache_id, ast_str_buffer(cache_id));
+		/* non-aggregate device state. */
+		new_device_state->stuff[0] = *eid;
+		new_device_state->eid = &new_device_state->stuff[0];
+		pos = (char *) &new_device_state->stuff[1];
 	} else {
-		/* no EID makes this an aggregate state */
-		ast_string_field_set(new_device_state, cache_id, device);
+		pos = (char *) &new_device_state->stuff[0];
 	}
 
-	ao2_ref(new_device_state, +1);
-	return new_device_state;
-}
-
-static enum ast_device_state get_aggregate_state(char *device)
-{
-	RAII_VAR(struct ao2_container *, cached, NULL, ao2_cleanup);
-	struct ast_devstate_aggregate aggregate;
-
-	ast_devstate_aggregate_init(&aggregate);
-
-	cached = stasis_cache_dump(ast_device_state_cache(), NULL);
-
-	ao2_callback_data(cached, OBJ_NODATA, devstate_change_aggregator_cb, &aggregate, device);
-
-	return ast_devstate_aggregate_result(&aggregate);
-}
-
-static int aggregate_state_changed(char *device, enum ast_device_state new_aggregate_state)
-{
-	RAII_VAR(struct stasis_message *, cached_aggregate_msg, NULL, ao2_cleanup);
-	struct ast_device_state_message *cached_aggregate_device_state;
+	strcpy(pos, device);/* Safe */
+	new_device_state->device = pos;
 
-	cached_aggregate_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device);
-	if (!cached_aggregate_msg) {
-		return 1;
-	}
+	new_device_state->state = state;
+	new_device_state->cachable = cachable;
 
-	cached_aggregate_device_state = stasis_message_data(cached_aggregate_msg);
-	if (cached_aggregate_device_state->state == new_aggregate_state) {
-		return 0;
-	}
-	return 1;
+	return new_device_state;
 }
 
-static void devstate_change_collector_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg)
+static void devstate_change_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg)
 {
-	enum ast_device_state aggregate_state;
-	char *device;
 	struct ast_device_state_message *device_state;
-	RAII_VAR(struct stasis_message *, new_aggregate_msg, NULL, ao2_cleanup);
-	RAII_VAR(struct ast_device_state_message *, new_aggregate_state, NULL, ao2_cleanup);
-
-	if (stasis_cache_update_type() == stasis_message_type(msg)) {
-		struct stasis_cache_update *update = stasis_message_data(msg);
-		if (!update->new_snapshot) {
-			return;
-		}
-		msg = update->new_snapshot;
-	}
 
 	if (ast_device_state_message_type() != stasis_message_type(msg)) {
 		return;
 	}
 
 	device_state = stasis_message_data(msg);
-
-	if (!device_state->eid) {
-		/* ignore aggregate messages */
+	if (device_state->cachable == AST_DEVSTATE_CACHABLE || !device_state->eid) {
+		/* Ignore cacheable and aggregate messages. */
 		return;
 	}
 
-	device = ast_strdupa(device_state->device);
-	ast_debug(1, "Processing device state change for '%s'\n", device);
-
-	if (device_state->cachable == AST_DEVSTATE_NOT_CACHABLE) {
-		/* if it's not cachable, there will be no aggregate state to get
-		 * and this should be passed through */
-		aggregate_state = device_state->state;
-	} else {
-
-		aggregate_state = get_aggregate_state(device);
-		ast_debug(1, "Aggregate devstate result is '%s' for '%s'\n",
-			ast_devstate2str(aggregate_state), device);
-
-		if (!aggregate_state_changed(device, aggregate_state)) {
-			/* No change since last reported device state */
-			ast_debug(1, "Aggregate state for device '%s' has not changed from '%s'\n",
-				device, ast_devstate2str(aggregate_state));
-			return;
-		}
-	}
-
-	ast_debug(1, "Aggregate state for device '%s' has changed to '%s'\n",
-		device, ast_devstate2str(aggregate_state));
-
-	ast_publish_device_state_full(device, aggregate_state, device_state->cachable, NULL);
+	/*
+	 * Non-cacheable device state aggregates are just the
+	 * device state republished as the aggregate.
+	 */
+	ast_publish_device_state_full(device_state->device, device_state->state,
+		device_state->cachable, NULL);
 }
 
 /*! \brief Initialize the device state engine in separate thread */
@@ -738,25 +656,30 @@ struct stasis_topic *ast_device_state_topic(const char *device)
 
 int ast_device_state_clear_cache(const char *device)
 {
-	RAII_VAR(struct stasis_message *, cached_msg, NULL, ao2_cleanup);
-	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+	struct stasis_message *cached_msg;
+	struct stasis_message *msg;
 
-	if (!(cached_msg = stasis_cache_get(ast_device_state_cache(),
-					    ast_device_state_message_type(), device))) {
+	cached_msg = stasis_cache_get_by_eid(ast_device_state_cache(),
+		ast_device_state_message_type(), device, &ast_eid_default);
+	if (!cached_msg) {
 		/* nothing to clear */
 		return -1;
 	}
 
 	msg = stasis_cache_clear_create(cached_msg);
-	stasis_publish(ast_device_state_topic(device), msg);
+	if (msg) {
+		stasis_publish(ast_device_state_topic(device), msg);
+	}
+	ao2_cleanup(msg);
+	ao2_cleanup(cached_msg);
 	return 0;
 }
 
 int ast_publish_device_state_full(
-			const char *device,
-			enum ast_device_state state,
-			enum ast_devstate_cache cachable,
-			struct ast_eid *eid)
+	const char *device,
+	enum ast_device_state state,
+	enum ast_devstate_cache cachable,
+	struct ast_eid *eid)
 {
 	RAII_VAR(struct ast_device_state_message *, device_state, NULL, ao2_cleanup);
 	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
@@ -769,7 +692,11 @@ int ast_publish_device_state_full(
 		return -1;
 	}
 
-	message = stasis_message_create(ast_device_state_message_type(), device_state);
+	message = stasis_message_create_full(ast_device_state_message_type(), device_state,
+		eid);
+	if (!message) {
+		return -1;
+	}
 
 	device_specific_topic = ast_device_state_topic(device);
 	if (!device_specific_topic) {
@@ -783,6 +710,7 @@ int ast_publish_device_state_full(
 static const char *device_state_get_id(struct stasis_message *message)
 {
 	struct ast_device_state_message *device_state;
+
 	if (ast_device_state_message_type() != stasis_message_type(message)) {
 		return NULL;
 	}
@@ -792,20 +720,124 @@ static const char *device_state_get_id(struct stasis_message *message)
 		return NULL;
 	}
 
-	return device_state->cache_id;
+	return device_state->device;
+}
+
+/*!
+ * \internal
+ * \brief Callback to publish the aggregate device state cache entry message.
+ * \since 12.2.0
+ *
+ * \param cache_topic Caching topic the aggregate message may be published over.
+ * \param aggregate The aggregate shapshot message to publish.
+ *
+ * \return Nothing
+ */
+static void device_state_aggregate_publish(struct stasis_topic *cache_topic, struct stasis_message *aggregate)
+{
+	const char *device;
+	struct stasis_topic *device_specific_topic;
+
+	device = device_state_get_id(aggregate);
+	if (!device) {
+		return;
+	}
+	device_specific_topic = ast_device_state_topic(device);
+	if (!device_specific_topic) {
+		return;
+	}
+
+	stasis_publish(device_specific_topic, aggregate);
+}
+
+/*!
+ * \internal
+ * \brief Callback to calculate the aggregate device state cache entry.
+ * \since 12.2.0
+ *
+ * \param entry Cache entry to calculate a new aggregate snapshot.
+ * \param new_snapshot The shapshot that is being updated.
+ *
+ * \note Return a ref bumped pointer from stasis_cache_entry_get_aggregate()
+ * if a new aggregate could not be calculated because of error.
+ *
+ * \return New aggregate-snapshot calculated on success.
+ * Caller has a reference on return.
+ */
+static struct stasis_message *device_state_aggregate_calc(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)
+{
+	struct stasis_message *aggregate_snapshot;
+	struct stasis_message *snapshot;
+	struct ast_device_state_message *device_state;
+	const char *device = NULL;
+	struct ast_devstate_aggregate aggregate;
+	int idx;
+
+	/* Determine the new aggregate device state. */
+	ast_devstate_aggregate_init(&aggregate);
+	snapshot = stasis_cache_entry_get_local(entry);
+	if (snapshot) {
+		device_state = stasis_message_data(snapshot);
+		device = device_state->device;
+		ast_devstate_aggregate_add(&aggregate, device_state->state);
+	}
+	for (idx = 0; ; ++idx) {
+		snapshot = stasis_cache_entry_get_remote(entry, idx);
+		if (!snapshot) {
+			break;
+		}
+
+		device_state = stasis_message_data(snapshot);
+		device = device_state->device;
+		ast_devstate_aggregate_add(&aggregate, device_state->state);
+	}
+
+	if (!device) {
+		/* There are no device states cached.  Delete the aggregate. */
+		return NULL;
+	}
+
+	snapshot = stasis_cache_entry_get_aggregate(entry);
+	if (snapshot) {
+		device_state = stasis_message_data(snapshot);
+		if (device_state->state == ast_devstate_aggregate_result(&aggregate)) {
+			/* Aggregate device state did not change. */
+			return ao2_bump(snapshot);
+		}
+	}
+
+	device_state = device_state_alloc(device, ast_devstate_aggregate_result(&aggregate),
+		AST_DEVSTATE_CACHABLE, NULL);
+	if (!device_state) {
+		/* Bummer.  We have to keep the old aggregate snapshot. */
+		return ao2_bump(snapshot);
+	}
+	aggregate_snapshot = stasis_message_create_full(ast_device_state_message_type(),
+		device_state, NULL);
+	ao2_cleanup(device_state);
+	if (!aggregate_snapshot) {
+		/* Bummer.  We have to keep the old aggregate snapshot. */
+		return ao2_bump(snapshot);
+	}
+
+	return aggregate_snapshot;
 }
 
 static void devstate_cleanup(void)
 {
 	devstate_message_sub = stasis_unsubscribe_and_join(devstate_message_sub);
-	ao2_cleanup(device_state_topic_all);
-	device_state_topic_all = NULL;
+	device_state_topic_cached = stasis_caching_unsubscribe_and_join(device_state_topic_cached);
+
 	ao2_cleanup(device_state_cache);
 	device_state_cache = NULL;
-	device_state_topic_cached = stasis_caching_unsubscribe_and_join(device_state_topic_cached);
-	STASIS_MESSAGE_TYPE_CLEANUP(ast_device_state_message_type);
+
 	ao2_cleanup(device_state_topic_pool);
 	device_state_topic_pool = NULL;
+
+	ao2_cleanup(device_state_topic_all);
+	device_state_topic_all = NULL;
+
+	STASIS_MESSAGE_TYPE_CLEANUP(ast_device_state_message_type);
 }
 
 int devstate_init(void)
@@ -817,25 +849,32 @@ int devstate_init(void)
 	}
 	device_state_topic_all = stasis_topic_create("ast_device_state_topic");
 	if (!device_state_topic_all) {
+		devstate_cleanup();
 		return -1;
 	}
-	device_state_cache = stasis_cache_create(device_state_get_id);
-	if (!device_state_cache) {
+	device_state_topic_pool = stasis_topic_pool_create(ast_device_state_topic_all());
+	if (!device_state_topic_pool) {
+		devstate_cleanup();
 		return -1;
 	}
-	device_state_topic_cached = stasis_caching_topic_create(device_state_topic_all, device_state_cache);
-	if (!device_state_topic_cached) {
+	device_state_cache = stasis_cache_create_full(device_state_get_id,
+		device_state_aggregate_calc, device_state_aggregate_publish);
+	if (!device_state_cache) {
+		devstate_cleanup();
 		return -1;
 	}
-	device_state_topic_pool = stasis_topic_pool_create(ast_device_state_topic_all());
-	if (!device_state_topic_pool) {
+	device_state_topic_cached = stasis_caching_topic_create(ast_device_state_topic_all(),
+		device_state_cache);
+	if (!device_state_topic_cached) {
+		devstate_cleanup();
 		return -1;
 	}
 
-	devstate_message_sub = stasis_subscribe(ast_device_state_topic_cached(), devstate_change_collector_cb, NULL);
-
+	devstate_message_sub = stasis_subscribe(ast_device_state_topic_all(),
+		devstate_change_cb, NULL);
 	if (!devstate_message_sub) {
-		ast_log(LOG_ERROR, "Failed to create subscription for the device state change collector\n");
+		ast_log(LOG_ERROR, "Failed to create subscription creating uncached device state aggregate events.\n");
+		devstate_cleanup();
 		return -1;
 	}
 
diff --git a/main/stasis_cache.c b/main/stasis_cache.c
index 1c5053915..22ace60af 100644
--- a/main/stasis_cache.c
+++ b/main/stasis_cache.c
@@ -36,6 +36,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/stasis_internal.h"
 #include "asterisk/stasis.h"
 #include "asterisk/utils.h"
+#include "asterisk/vector.h"
 
 #ifdef LOW_MEMORY
 #define NUM_CACHE_BUCKETS 17
@@ -47,6 +48,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 struct stasis_cache {
 	struct ao2_container *entries;
 	snapshot_get_id id_fn;
+	cache_aggregate_calc_fn aggregate_calc_fn;
+	cache_aggregate_publish_fn aggregate_publish_fn;
 };
 
 /*! \internal */
@@ -124,29 +127,53 @@ struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_c
 	return NULL;
 }
 
-struct cache_entry {
+struct cache_entry_key {
 	struct stasis_message_type *type;
-	char *id;
-	struct stasis_message *snapshot;
+	const char *id;
+};
+
+struct stasis_cache_entry {
+	struct cache_entry_key key;
+	/*! Aggregate snapshot of the stasis cache. */
+	struct stasis_message *aggregate;
+	/*! Local entity snapshot of the stasis event. */
+	struct stasis_message *local;
+	/*! Remote entity snapshots of the stasis event. */
+	AST_VECTOR(, struct stasis_message *) remote;
 };
 
 static void cache_entry_dtor(void *obj)
 {
-	struct cache_entry *entry = obj;
-	ao2_cleanup(entry->type);
-	entry->type = NULL;
-	ast_free(entry->id);
-	entry->id = NULL;
-	ao2_cleanup(entry->snapshot);
-	entry->snapshot = NULL;
+	struct stasis_cache_entry *entry = obj;
+	size_t idx;
+
+	ao2_cleanup(entry->key.type);
+	entry->key.type = NULL;
+	ast_free((char *) entry->key.id);
+	entry->key.id = NULL;
+
+	ao2_cleanup(entry->aggregate);
+	entry->aggregate = NULL;
+	ao2_cleanup(entry->local);
+	entry->local = NULL;
+
+	for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
+		struct stasis_message *remote;
+
+		remote = AST_VECTOR_GET(&entry->remote, idx);
+		ao2_cleanup(remote);
+	}
+	AST_VECTOR_FREE(&entry->remote);
 }
 
-static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
+static struct stasis_cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
 {
-	RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
+	struct stasis_cache_entry *entry;
+	int is_remote;
 
 	ast_assert(type != NULL);
 	ast_assert(id != NULL);
+	ast_assert(snapshot != NULL);
 
 	entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
 		AO2_ALLOC_OPT_LOCK_NOLOCK);
@@ -154,173 +181,528 @@ static struct cache_entry *cache_entry_create(struct stasis_message_type *type,
 		return NULL;
 	}
 
-	entry->id = ast_strdup(id);
-	if (!entry->id) {
+	entry->key.id = ast_strdup(id);
+	if (!entry->key.id) {
+		ao2_cleanup(entry);
+		return NULL;
+	}
+	entry->key.type = ao2_bump(type);
+
+	is_remote = ast_eid_cmp(&ast_eid_default, stasis_message_eid(snapshot)) ? 1 : 0;
+	if (AST_VECTOR_INIT(&entry->remote, is_remote)) {
+		ao2_cleanup(entry);
 		return NULL;
 	}
 
-	ao2_ref(type, +1);
-	entry->type = type;
-	if (snapshot != NULL) {
-		ao2_ref(snapshot, +1);
-		entry->snapshot = snapshot;
+	if (is_remote) {
+		if (AST_VECTOR_APPEND(&entry->remote, snapshot)) {
+			ao2_cleanup(entry);
+			return NULL;
+		}
+	} else {
+		entry->local = snapshot;
 	}
+	ao2_bump(snapshot);
 
-	ao2_ref(entry, +1);
 	return entry;
 }
 
 static int cache_entry_hash(const void *obj, int flags)
 {
-	const struct cache_entry *entry = obj;
+	const struct stasis_cache_entry *object;
+	const struct cache_entry_key *key;
 	int hash = 0;
 
-	ast_assert(!(flags & OBJ_KEY));
+	switch (flags & OBJ_SEARCH_MASK) {
+	case OBJ_SEARCH_KEY:
+		key = obj;
+		break;
+	case OBJ_SEARCH_OBJECT:
+		object = obj;
+		key = &object->key;
+		break;
+	default:
+		/* Hash can only work on something with a full key. */
+		ast_assert(0);
+		return 0;
+	}
 
-	hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
-	hash += ast_hashtab_hash_string(entry->id);
+	hash += ast_hashtab_hash_string(stasis_message_type_name(key->type));
+	hash += ast_hashtab_hash_string(key->id);
 	return hash;
 }
 
 static int cache_entry_cmp(void *obj, void *arg, int flags)
 {
-	const struct cache_entry *left = obj;
-	const struct cache_entry *right = arg;
+	const struct stasis_cache_entry *object_left = obj;
+	const struct stasis_cache_entry *object_right = arg;
+	const struct cache_entry_key *right_key = obj;
+	int cmp;
+
+	switch (flags & OBJ_SEARCH_MASK) {
+	case OBJ_SEARCH_OBJECT:
+		right_key = &object_right->key;
+		/* Fall through */
+	case OBJ_SEARCH_KEY:
+		cmp = object_left->key.type != right_key->type
+			|| strcmp(object_left->key.id, right_key->id);
+		break;
+	case OBJ_SEARCH_PARTIAL_KEY:
+		/* Not supported by container */
+		ast_assert(0);
+		cmp = -1;
+		break;
+	default:
+		/*
+		 * What arg points to is specific to this traversal callback
+		 * and has no special meaning to astobj2.
+		 */
+		cmp = 0;
+		break;
+	}
+	if (cmp) {
+		return 0;
+	}
+	/*
+	 * At this point the traversal callback is identical to a sorted
+	 * container.
+	 */
+	return CMP_MATCH;
+}
+
+static void cache_dtor(void *obj)
+{
+	struct stasis_cache *cache = obj;
 
-	ast_assert(!(flags & OBJ_KEY));
+	ao2_cleanup(cache->entries);
+	cache->entries = NULL;
+}
 
-	if (left->type == right->type && strcmp(left->id, right->id) == 0) {
-		return CMP_MATCH | CMP_STOP;
+struct stasis_cache *stasis_cache_create_full(snapshot_get_id id_fn,
+	cache_aggregate_calc_fn aggregate_calc_fn,
+	cache_aggregate_publish_fn aggregate_publish_fn)
+{
+	struct stasis_cache *cache;
+
+	cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
+		AO2_ALLOC_OPT_LOCK_NOLOCK);
+	if (!cache) {
+		return NULL;
 	}
 
-	return 0;
+	cache->entries = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
+		NUM_CACHE_BUCKETS, cache_entry_hash, NULL, cache_entry_cmp);
+	if (!cache->entries) {
+		ao2_cleanup(cache);
+		return NULL;
+	}
+
+	cache->id_fn = id_fn;
+	cache->aggregate_calc_fn = aggregate_calc_fn;
+	cache->aggregate_publish_fn = aggregate_publish_fn;
+
+	return cache;
 }
 
-static void cache_dtor(void *obj)
+struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
 {
-        struct stasis_cache *cache = obj;
+	return stasis_cache_create_full(id_fn, NULL, NULL);
+}
 
-        ao2_cleanup(cache->entries);
-        cache->entries = NULL;
+struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry)
+{
+	return entry->aggregate;
 }
 
-struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
+struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry)
 {
-        RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
+	return entry->local;
+}
 
-        cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
-		AO2_ALLOC_OPT_LOCK_NOLOCK);
-        if (!cache) {
-                return NULL;
-        }
+struct stasis_message *stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx)
+{
+	if (idx < AST_VECTOR_SIZE(&entry->remote)) {
+		return AST_VECTOR_GET(&entry->remote, idx);
+	}
+	return NULL;
+}
 
-        cache->entries = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash,
-                cache_entry_cmp);
-        if (!cache->entries) {
-                return NULL;
-        }
+/*!
+ * \internal
+ * \brief Find the cache entry in the cache entries container.
+ *
+ * \param entries Container of cached entries.
+ * \param type Type of message to retrieve the cache entry.
+ * \param id Identity of the snapshot to retrieve the cache entry.
+ *
+ * \note The entries container is already locked.
+ *
+ * \retval Cache-entry on success.
+ * \retval NULL Not in cache.
+ */
+static struct stasis_cache_entry *cache_find(struct ao2_container *entries, struct stasis_message_type *type, const char *id)
+{
+	struct cache_entry_key search_key;
+
+	search_key.type = type;
+	search_key.id = id;
+	return ao2_find(entries, &search_key, OBJ_SEARCH_KEY | OBJ_NOLOCK);
+}
+
+/*!
+ * \internal
+ * \brief Remove the stasis snapshot in the cache entry determined by eid.
+ *
+ * \param entries Container of cached entries.
+ * \param cached_entry The entry to remove the snapshot from.
+ * \param eid Which snapshot in the cached entry.
+ *
+ * \note The entries container is already locked.
+ *
+ * \return Previous stasis entry snapshot.
+ */
+static struct stasis_message *cache_remove(struct ao2_container *entries, struct stasis_cache_entry *cached_entry, const struct ast_eid *eid)
+{
+	struct stasis_message *old_snapshot;
+	int is_remote;
+
+	is_remote = ast_eid_cmp(eid, &ast_eid_default);
+	if (!is_remote) {
+		old_snapshot = cached_entry->local;
+		cached_entry->local = NULL;
+	} else {
+		int idx;
+
+		old_snapshot = NULL;
+		for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
+			struct stasis_message *cur;
+
+			cur = AST_VECTOR_GET(&cached_entry->remote, idx);
+			if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
+				old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
+				break;
+			}
+		}
+	}
+
+	if (!cached_entry->local && !AST_VECTOR_SIZE(&cached_entry->remote)) {
+		ao2_unlink_flags(entries, cached_entry, OBJ_NOLOCK);
+	}
+
+	return old_snapshot;
+}
 
-        cache->id_fn = id_fn;
+/*!
+ * \internal
+ * \brief Update the stasis snapshot in the cache entry determined by eid.
+ *
+ * \param cached_entry The entry to remove the snapshot from.
+ * \param eid Which snapshot in the cached entry.
+ * \param new_snapshot Snapshot to replace the old snapshot.
+ *
+ * \return Previous stasis entry snapshot.
+ */
+static struct stasis_message *cache_udpate(struct stasis_cache_entry *cached_entry, const struct ast_eid *eid, struct stasis_message *new_snapshot)
+{
+	struct stasis_message *old_snapshot;
+	int is_remote;
+	int idx;
+
+	is_remote = ast_eid_cmp(eid, &ast_eid_default);
+	if (!is_remote) {
+		old_snapshot = cached_entry->local;
+		cached_entry->local = ao2_bump(new_snapshot);
+		return old_snapshot;
+	}
+
+	old_snapshot = NULL;
+	for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
+		struct stasis_message *cur;
 
-        ao2_ref(cache, +1);
-        return cache;
+		cur = AST_VECTOR_GET(&cached_entry->remote, idx);
+		if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
+			old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
+			break;
+		}
+	}
+	if (!AST_VECTOR_APPEND(&cached_entry->remote, new_snapshot)) {
+		ao2_bump(new_snapshot);
+	}
+
+	return old_snapshot;
 }
 
-static struct stasis_message *cache_put(struct stasis_cache *cache,
-	struct stasis_message_type *type, const char *id,
+struct cache_put_snapshots {
+	/*! Old cache eid snapshot. */
+	struct stasis_message *old;
+	/*! Old cache aggregate snapshot. */
+	struct stasis_message *aggregate_old;
+	/*! New cache aggregate snapshot. */
+	struct stasis_message *aggregate_new;
+};
+
+static struct cache_put_snapshots cache_put(struct stasis_cache *cache,
+	struct stasis_message_type *type, const char *id, const struct ast_eid *eid,
 	struct stasis_message *new_snapshot)
 {
-	RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
-	RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
-	struct stasis_message *old_snapshot = NULL;
+	struct stasis_cache_entry *cached_entry;
+	struct cache_put_snapshots snapshots;
 
 	ast_assert(cache->entries != NULL);
+	ast_assert(eid != NULL);/* Aggregate snapshots not allowed to be put directly. */
 	ast_assert(new_snapshot == NULL ||
 		type == stasis_message_type(new_snapshot));
 
-	new_entry = cache_entry_create(type, id, new_snapshot);
+	memset(&snapshots, 0, sizeof(snapshots));
+
+	ao2_wrlock(cache->entries);
+
+	cached_entry = cache_find(cache->entries, type, id);
 
-	if (new_snapshot == NULL) {
-		/* Remove entry from cache */
-		cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_UNLINK);
+	/* Update the eid snapshot. */
+	if (!new_snapshot) {
+		/* Remove snapshot from cache */
 		if (cached_entry) {
-			old_snapshot = cached_entry->snapshot;
-			cached_entry->snapshot = NULL;
+			snapshots.old = cache_remove(cache->entries, cached_entry, eid);
 		}
+	} else if (cached_entry) {
+		/* Update snapshot in cache */
+		snapshots.old = cache_udpate(cached_entry, eid, new_snapshot);
 	} else {
-		/* Insert/update cache */
-		SCOPED_AO2LOCK(lock, cache->entries);
-
-		cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_NOLOCK);
+		/* Insert into the cache */
+		cached_entry = cache_entry_create(type, id, new_snapshot);
 		if (cached_entry) {
-			/* Update cache. Because objects are moving, no need to update refcounts. */
-			old_snapshot = cached_entry->snapshot;
-			cached_entry->snapshot = new_entry->snapshot;
-			new_entry->snapshot = NULL;
-		} else {
-			/* Insert into the cache */
-			ao2_link_flags(cache->entries, new_entry, OBJ_NOLOCK);
+			ao2_link_flags(cache->entries, cached_entry, OBJ_NOLOCK);
 		}
+	}
 
+	/* Update the aggregate snapshot. */
+	if (cache->aggregate_calc_fn && cached_entry) {
+		snapshots.aggregate_new = cache->aggregate_calc_fn(cached_entry, new_snapshot);
+		snapshots.aggregate_old = cached_entry->aggregate;
+		cached_entry->aggregate = ao2_bump(snapshots.aggregate_new);
 	}
 
-	return old_snapshot;
+	ao2_unlock(cache->entries);
+
+	ao2_cleanup(cached_entry);
+	return snapshots;
 }
 
-struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
+/*!
+ * \internal
+ * \brief Dump all entity snapshots in the cache entry into the given container.
+ *
+ * \param snapshots Container to put all snapshots in the cache entry.
+ * \param entry Cache entry to use.
+ *
+ * \retval 0 on success.
+ * \retval non-zero on error.
+ */
+static int cache_entry_dump(struct ao2_container *snapshots, const struct stasis_cache_entry *entry)
+{
+	int idx;
+	int err = 0;
+
+	ast_assert(snapshots != NULL);
+	ast_assert(entry != NULL);
+
+	/* The aggregate snapshot is not a snapshot from an entity. */
+
+	if (entry->local) {
+		err |= !ao2_link(snapshots, entry->local);
+	}
+
+	for (idx = 0; !err && idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
+		struct stasis_message *snapshot;
+
+		snapshot = AST_VECTOR_GET(&entry->remote, idx);
+		err |= !ao2_link(snapshots, snapshot);
+	}
+
+	return err;
+}
+
+struct ao2_container *stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
 {
-	RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
-	RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
+	struct stasis_cache_entry *cached_entry;
+	struct ao2_container *found;
 
+	ast_assert(cache != NULL);
 	ast_assert(cache->entries != NULL);
+	ast_assert(type != NULL);
+	ast_assert(id != NULL);
 
-	search_entry = cache_entry_create(type, id, NULL);
-	if (search_entry == NULL) {
+	found = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
+	if (!found) {
 		return NULL;
 	}
 
-	cached_entry = ao2_find(cache->entries, search_entry, OBJ_POINTER);
-	if (cached_entry == NULL) {
-		return NULL;
+	ao2_rdlock(cache->entries);
+
+	cached_entry = cache_find(cache->entries, type, id);
+	if (cached_entry && cache_entry_dump(found, cached_entry)) {
+		ao2_cleanup(found);
+		found = NULL;
+	}
+
+	ao2_unlock(cache->entries);
+
+	ao2_cleanup(cached_entry);
+	return found;
+}
+
+/*!
+ * \internal
+ * \brief Retrieve an item from the cache entry for a specific eid.
+ *
+ * \param entry Cache entry to use.
+ * \param eid Specific entity id to retrieve.  NULL for aggregate.
+ *
+ * \note The returned snapshot has not had its reference bumped.
+ *
+ * \retval Snapshot from the cache.
+ * \retval \c NULL if snapshot is not found.
+ */
+static struct stasis_message *cache_entry_by_eid(const struct stasis_cache_entry *entry, const struct ast_eid *eid)
+{
+	int is_remote;
+	int idx;
+
+	if (!eid) {
+		/* Get aggregate. */
+		return entry->aggregate;
+	}
+
+	/* Get snapshot with specific eid. */
+	is_remote = ast_eid_cmp(eid, &ast_eid_default);
+	if (!is_remote) {
+		return entry->local;
 	}
 
-	ast_assert(cached_entry->snapshot != NULL);
-	ao2_ref(cached_entry->snapshot, +1);
-	return cached_entry->snapshot;
+	for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
+		struct stasis_message *cur;
+
+		cur = AST_VECTOR_GET(&entry->remote, idx);
+		if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
+			return cur;
+		}
+	}
+
+	return NULL;
+}
+
+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)
+{
+	struct stasis_cache_entry *cached_entry;
+	struct stasis_message *snapshot = NULL;
+
+	ast_assert(cache != NULL);
+	ast_assert(cache->entries != NULL);
+	ast_assert(type != NULL);
+	ast_assert(id != NULL);
+
+	ao2_rdlock(cache->entries);
+
+	cached_entry = cache_find(cache->entries, type, id);
+	if (cached_entry) {
+		snapshot = cache_entry_by_eid(cached_entry, eid);
+		ao2_bump(snapshot);
+	}
+
+	ao2_unlock(cache->entries);
+
+	ao2_cleanup(cached_entry);
+	return snapshot;
+}
+
+struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
+{
+	return stasis_cache_get_by_eid(cache, type, id, &ast_eid_default);
 }
 
 struct cache_dump_data {
-	struct ao2_container *cached;
+	struct ao2_container *container;
 	struct stasis_message_type *type;
+	const struct ast_eid *eid;
 };
 
-static int cache_dump_cb(void *obj, void *arg, int flags)
+static int cache_dump_by_eid_cb(void *obj, void *arg, int flags)
 {
 	struct cache_dump_data *cache_dump = arg;
-	struct cache_entry *entry = obj;
+	struct stasis_cache_entry *entry = obj;
+
+	if (!cache_dump->type || entry->key.type == cache_dump->type) {
+		struct stasis_message *snapshot;
 
-	if (!cache_dump->type || entry->type == cache_dump->type) {
-		ao2_link(cache_dump->cached, entry->snapshot);
+		snapshot = cache_entry_by_eid(entry, cache_dump->eid);
+		if (snapshot) {
+			if (!ao2_link(cache_dump->container, snapshot)) {
+				ao2_cleanup(cache_dump->container);
+				cache_dump->container = NULL;
+				return CMP_STOP;
+			}
+		}
 	}
 
 	return 0;
 }
 
+struct ao2_container *stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
+{
+	struct cache_dump_data cache_dump;
+
+	ast_assert(cache != NULL);
+	ast_assert(cache->entries != NULL);
+
+	cache_dump.eid = eid;
+	cache_dump.type = type;
+	cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
+	if (!cache_dump.container) {
+		return NULL;
+	}
+
+	ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_by_eid_cb, &cache_dump);
+	return cache_dump.container;
+}
+
 struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
 {
+	return stasis_cache_dump_by_eid(cache, type, &ast_eid_default);
+}
+
+static int cache_dump_all_cb(void *obj, void *arg, int flags)
+{
+	struct cache_dump_data *cache_dump = arg;
+	struct stasis_cache_entry *entry = obj;
+
+	if (!cache_dump->type || entry->key.type == cache_dump->type) {
+		if (cache_entry_dump(cache_dump->container, entry)) {
+			ao2_cleanup(cache_dump->container);
+			cache_dump->container = NULL;
+			return CMP_STOP;
+		}
+	}
+
+	return 0;
+}
+
+struct ao2_container *stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type)
+{
 	struct cache_dump_data cache_dump;
 
+	ast_assert(cache != NULL);
 	ast_assert(cache->entries != NULL);
 
+	cache_dump.eid = NULL;
 	cache_dump.type = type;
-	cache_dump.cached = ao2_container_alloc_options(
-		AO2_ALLOC_OPT_LOCK_NOLOCK, 1, NULL, NULL);
-	if (!cache_dump.cached) {
+	cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
+	if (!cache_dump.container) {
 		return NULL;
 	}
 
-	ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump);
-	return cache_dump.cached;
+	ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_all_cb, &cache_dump);
+	return cache_dump.container;
 }
 
 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
@@ -380,9 +762,13 @@ static struct stasis_message *update_create(struct stasis_message *old_snapshot,
 static void caching_topic_exec(void *data, struct stasis_subscription *sub,
 	struct stasis_message *message)
 {
-	RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
+	struct stasis_caching_topic *caching_topic_needs_unref;
 	struct stasis_caching_topic *caching_topic = data;
-	const char *id = NULL;
+	struct stasis_message *msg;
+	struct stasis_message *msg_put;
+	struct stasis_message_type *msg_type;
+	const struct ast_eid *msg_eid;
+	const char *msg_id;
 
 	ast_assert(caching_topic != NULL);
 	ast_assert(caching_topic->topic != NULL);
@@ -391,50 +777,62 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub,
 
 	if (stasis_subscription_final_message(sub, message)) {
 		caching_topic_needs_unref = caching_topic;
+	} else {
+		caching_topic_needs_unref = NULL;
 	}
 
-	/* Handle cache clear event */
-	if (stasis_cache_clear_type() == stasis_message_type(message)) {
-		RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
-		RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
-		struct stasis_message *clear_msg = stasis_message_data(message);
-		const char *clear_id = caching_topic->cache->id_fn(clear_msg);
-		struct stasis_message_type *clear_type = stasis_message_type(clear_msg);
+	msg_type = stasis_message_type(message);
+	if (stasis_cache_clear_type() == msg_type) {
+		/* Cache clear event. */
+		msg_put = NULL;
+		msg = stasis_message_data(message);
+		msg_type = stasis_message_type(msg);
+	} else {
+		/* Normal cache update event. */
+		msg_put = message;
+		msg = message;
+	}
+	ast_assert(msg_type != NULL);
 
-		ast_assert(clear_type != NULL);
+	msg_eid = stasis_message_eid(msg);/* msg_eid is NULL for aggregate message. */
+	msg_id = caching_topic->cache->id_fn(msg);
+	if (msg_id && msg_eid) {
+		struct stasis_message *update;
+		struct cache_put_snapshots snapshots;
 
-		if (clear_id) {
-			old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL);
-			if (old_snapshot) {
-				update = update_create(old_snapshot, NULL);
+		/* Update the cache */
+		snapshots = cache_put(caching_topic->cache, msg_type, msg_id, msg_eid, msg_put);
+		if (snapshots.old || msg_put) {
+			update = update_create(snapshots.old, msg_put);
+			if (update) {
 				stasis_publish(caching_topic->topic, update);
-				return;
 			}
-
+			ao2_cleanup(update);
+		} else {
 			ast_log(LOG_ERROR,
 				"Attempting to remove an item from the %s cache that isn't there: %s %s\n",
-				stasis_topic_name(caching_topic->topic), stasis_message_type_name(clear_type), clear_id);
-			return;
+				stasis_topic_name(caching_topic->topic),
+				stasis_message_type_name(msg_type), msg_id);
 		}
-	}
-
-	id = caching_topic->cache->id_fn(message);
-	if (id == NULL) {
-		/* Object isn't cached; discard */
-	} else {
-		/* Update the cache */
-		RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
-		RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
 
-		old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message);
-
-		update = update_create(old_snapshot, message);
-		if (update == NULL) {
-			return;
+		if (snapshots.aggregate_old != snapshots.aggregate_new) {
+			if (snapshots.aggregate_new && caching_topic->cache->aggregate_publish_fn) {
+				caching_topic->cache->aggregate_publish_fn(caching_topic->original_topic,
+					snapshots.aggregate_new);
+			}
+			update = update_create(snapshots.aggregate_old, snapshots.aggregate_new);
+			if (update) {
+				stasis_publish(caching_topic->topic, update);
+			}
+			ao2_cleanup(update);
 		}
 
-		stasis_publish(caching_topic->topic, update);
+		ao2_cleanup(snapshots.old);
+		ao2_cleanup(snapshots.aggregate_old);
+		ao2_cleanup(snapshots.aggregate_new);
 	}
+
+	ao2_cleanup(caching_topic_needs_unref);
 }
 
 struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)
diff --git a/main/stasis_message.c b/main/stasis_message.c
index 240845aff..1db2ae97a 100644
--- a/main/stasis_message.c
+++ b/main/stasis_message.c
@@ -53,7 +53,7 @@ static void message_type_dtor(void *obj)
 struct stasis_message_type *stasis_message_type_create(const char *name,
 	struct stasis_message_vtable *vtable)
 {
-	RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
+	struct stasis_message_type *type;
 
 	type = ao2_alloc(sizeof(*type), message_type_dtor);
 	if (!type) {
@@ -66,11 +66,11 @@ struct stasis_message_type *stasis_message_type_create(const char *name,
 
 	type->name = ast_strdup(name);
 	if (!type->name) {
+		ao2_cleanup(type);
 		return NULL;
 	}
 	type->vtable = vtable;
 
-	ao2_ref(type, +1);
 	return type;
 }
 
@@ -85,8 +85,12 @@ struct stasis_message {
 	struct timeval timestamp;
 	/*! Type of the message */
 	struct stasis_message_type *type;
+	/*! Where this message originated.  NULL if aggregate message. */
+	const struct ast_eid *eid_ptr;
 	/*! Message content */
 	void *data;
+	/*! Where this message originated. */
+	struct ast_eid eid;
 };
 
 static void stasis_message_dtor(void *obj)
@@ -96,9 +100,9 @@ static void stasis_message_dtor(void *obj)
 	ao2_cleanup(message->data);
 }
 
-struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data)
+struct stasis_message *stasis_message_create_full(struct stasis_message_type *type, void *data, const struct ast_eid *eid)
 {
-	RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
+	struct stasis_message *message;
 
 	if (type == NULL || data == NULL) {
 		return NULL;
@@ -114,11 +118,27 @@ struct stasis_message *stasis_message_create(struct stasis_message_type *type, v
 	message->type = type;
 	ao2_ref(data, +1);
 	message->data = data;
+	if (eid) {
+		message->eid_ptr = &message->eid;
+		message->eid = *eid;
+	}
 
-	ao2_ref(message, +1);
 	return message;
 }
 
+struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data)
+{
+	return stasis_message_create_full(type, data, &ast_eid_default);
+}
+
+const struct ast_eid *stasis_message_eid(const struct stasis_message *msg)
+{
+	if (msg == NULL) {
+		return NULL;
+	}
+	return msg->eid_ptr;
+}
+
 struct stasis_message_type *stasis_message_type(const struct stasis_message *msg)
 {
 	if (msg == NULL) {