diff options
author | Kinsey Moore <kmoore@digium.com> | 2013-04-16 15:33:59 +0000 |
---|---|---|
committer | Kinsey Moore <kmoore@digium.com> | 2013-04-16 15:33:59 +0000 |
commit | 191cf99ae1f821dec98199931fc775fc0716d27c (patch) | |
tree | 210b5ebfaf3f30fa06c666e8d99e93e9918253a5 /tests/test_devicestate.c | |
parent | c1ae5dc49be01a567b403306ee4b560391293f67 (diff) |
Move device state distribution to Stasis-core
In the move from Asterisk's event system to Stasis, this makes
distributed device state aggregation always-on, removes unnecessary
task processors where possible, and collapses aggregate and
non-aggregate states into a single cache for ease of retrieval. This
also removes an intermediary step in device state aggregation.
Review: https://reviewboard.asterisk.org/r/2389/
(closes issue ASTERISK-21101)
Patch-by: Kinsey Moore <kmoore@digium.com>
git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@385860 65c4cc65-6c06-0410-ace0-fbb531ad65f3
Diffstat (limited to 'tests/test_devicestate.c')
-rw-r--r-- | tests/test_devicestate.c | 229 |
1 files changed, 229 insertions, 0 deletions
diff --git a/tests/test_devicestate.c b/tests/test_devicestate.c index f5f209b9e..233b6e70a 100644 --- a/tests/test_devicestate.c +++ b/tests/test_devicestate.c @@ -39,7 +39,9 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") #include "asterisk/test.h" #include "asterisk/devicestate.h" #include "asterisk/pbx.h" +#include "asterisk/stasis_message_router.h" +#define UNIT_TEST_DEVICE_IDENTIFIER "unit_test_device_identifier" /* These arrays are the result of the 'core show device2extenstate' output. */ static int combined_results[] = { @@ -274,14 +276,241 @@ AST_TEST_DEFINE(device2extenstate_test) return res; } +struct consumer { + ast_mutex_t lock; + ast_cond_t out; + int already_out; + enum ast_device_state state; + enum ast_device_state aggregate_state; + int sig_on_non_aggregate_state; +}; + +static void consumer_dtor(void *obj) { + struct consumer *consumer = obj; + + ast_mutex_destroy(&consumer->lock); + ast_cond_destroy(&consumer->out); +} + +static struct consumer *consumer_create(void) { + RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup); + + consumer = ao2_alloc(sizeof(*consumer), consumer_dtor); + + if (!consumer) { + return NULL; + } + + ast_mutex_init(&consumer->lock); + ast_cond_init(&consumer->out, NULL); + consumer->sig_on_non_aggregate_state = 0; + + ao2_ref(consumer, +1); + return consumer; +} + +static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message) +{ + struct consumer *consumer = data; + RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup); + struct stasis_cache_update *cache_update = stasis_message_data(message); + struct ast_device_state_message *device_state; + SCOPED_MUTEX(lock, &consumer->lock); + + if (!cache_update->new_snapshot) { + return; + } + + device_state = stasis_message_data(cache_update->new_snapshot); + + if (strcmp(device_state->device, UNIT_TEST_DEVICE_IDENTIFIER)) { + /* not a device state we're interested in */ + return; + } + + if (device_state->eid) { + consumer->state = device_state->state; + if (consumer->sig_on_non_aggregate_state) { + consumer->sig_on_non_aggregate_state = 0; + consumer->already_out = 1; + ast_cond_signal(&consumer->out); + } + } else { + consumer->aggregate_state = device_state->state; + consumer->already_out = 1; + ast_cond_signal(&consumer->out); + } +} + +static void consumer_finalize(void *data, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *message) +{ + struct consumer *consumer = data; + + if (stasis_subscription_final_message(sub, message)) { + ao2_cleanup(consumer); + } +} + +static void consumer_wait_for(struct consumer *consumer) +{ + int res; + struct timeval start = ast_tvnow(); + struct timespec end = { + .tv_sec = start.tv_sec + 10, + .tv_nsec = start.tv_usec * 1000 + }; + + SCOPED_MUTEX(lock, &consumer->lock); + + if (consumer->already_out) { + consumer->already_out = 0; + } + + while(1) { + res = ast_cond_timedwait(&consumer->out, &consumer->lock, &end); + if (!res || res == ETIMEDOUT) { + break; + } + } + consumer->already_out = 0; +} + +static int remove_device_states_cb(void *obj, void *arg, int flags) +{ + RAII_VAR(struct stasis_message *, msg, obj, ao2_cleanup); + struct ast_device_state_message *device_state = stasis_message_data(msg); + if (strcmp(UNIT_TEST_DEVICE_IDENTIFIER, device_state->device)) { + msg = NULL; + return 0; + } + + msg = stasis_cache_clear_create(ast_device_state_message_type(), device_state->cache_id); + /* topic guaranteed to have been created by this point */ + stasis_publish(ast_device_state_topic(device_state->device), msg); + return 0; +} + +static void cache_cleanup(int unused) +{ + RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup); + /* remove all device states created during this test */ + cache_dump = stasis_cache_dump(ast_device_state_topic_cached(), NULL); + if (!cache_dump) { + return; + } + ao2_callback(cache_dump, 0, remove_device_states_cb, NULL); +} + +AST_TEST_DEFINE(device_state_aggregation_test) +{ + RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup); + RAII_VAR(struct stasis_message_router *, device_msg_router, NULL, stasis_message_router_unsubscribe); + RAII_VAR(struct ast_eid *, foreign_eid, NULL, ast_free); + RAII_VAR(int, cleanup_cache, 0, cache_cleanup); + int res; + struct ast_device_state_message *device_state; + struct stasis_message *msg; + + switch (cmd) { + case TEST_INIT: + info->name = "device_state_aggregation_test"; + info->category = "/main/devicestate/"; + info->summary = "Tests message routing and aggregation through the Stasis device state system."; + info->description = + "Verifies that the device state system passes " + "messages appropriately, that the aggregator is " + "working properly, that the aggregate results match " + "the expected combined devstate, and that the cached " + "aggregate devstate is correct."; + return AST_TEST_NOT_RUN; + case TEST_EXECUTE: + break; + } + + foreign_eid = ast_malloc(sizeof(*foreign_eid)); + ast_test_validate(test, NULL != foreign_eid); + memset(foreign_eid, 0xFF, sizeof(*foreign_eid)); + + consumer = consumer_create(); + ast_test_validate(test, NULL != consumer); + + device_msg_router = stasis_message_router_create(stasis_caching_get_topic(ast_device_state_topic_cached())); + ast_test_validate(test, NULL != device_msg_router); + + ao2_ref(consumer, +1); + res = stasis_message_router_add(device_msg_router, stasis_cache_update_type(), consumer_exec, consumer); + ast_test_validate(test, !res); + + res = stasis_message_router_add(device_msg_router, stasis_subscription_change_type(), consumer_finalize, consumer); + ast_test_validate(test, !res); + + /* push local state */ + ast_publish_device_state(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE); + + consumer_wait_for(consumer); + ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->state); + ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->aggregate_state); + + msg = stasis_cache_get(ast_device_state_topic_cached(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER); + device_state = stasis_message_data(msg); + ast_test_validate(test, AST_DEVICE_NOT_INUSE == device_state->state); + ao2_cleanup(msg); + msg = NULL; + + /* push remote state */ + /* this will not produce a new aggregate state message since the aggregate state does not change */ + consumer->sig_on_non_aggregate_state = 1; + ast_publish_device_state_full(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE, foreign_eid); + + consumer_wait_for(consumer); + ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->state); + ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->aggregate_state); + + msg = stasis_cache_get(ast_device_state_topic_cached(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER); + device_state = stasis_message_data(msg); + ast_test_validate(test, AST_DEVICE_NOT_INUSE == device_state->state); + ao2_cleanup(msg); + msg = NULL; + + /* push remote state different from local state */ + ast_publish_device_state_full(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_INUSE, AST_DEVSTATE_CACHABLE, foreign_eid); + + consumer_wait_for(consumer); + ast_test_validate(test, AST_DEVICE_INUSE == consumer->state); + ast_test_validate(test, AST_DEVICE_INUSE == consumer->aggregate_state); + + msg = stasis_cache_get(ast_device_state_topic_cached(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER); + device_state = stasis_message_data(msg); + ast_test_validate(test, AST_DEVICE_INUSE == device_state->state); + ao2_cleanup(msg); + msg = NULL; + + /* push local state that will cause aggregated state different from local non-aggregate state */ + ast_publish_device_state(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_RINGING, AST_DEVSTATE_CACHABLE); + + consumer_wait_for(consumer); + ast_test_validate(test, AST_DEVICE_RINGING == consumer->state); + ast_test_validate(test, AST_DEVICE_RINGINUSE == consumer->aggregate_state); + + msg = stasis_cache_get(ast_device_state_topic_cached(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER); + device_state = stasis_message_data(msg); + ast_test_validate(test, AST_DEVICE_RINGINUSE == device_state->state); + ao2_cleanup(msg); + msg = NULL; + + return AST_TEST_PASS; +} + static int unload_module(void) { AST_TEST_UNREGISTER(device2extenstate_test); + AST_TEST_UNREGISTER(device_state_aggregation_test); return 0; } static int load_module(void) { + AST_TEST_REGISTER(device_state_aggregation_test); AST_TEST_REGISTER(device2extenstate_test); return AST_MODULE_LOAD_SUCCESS; } |