[asterisk-commits] file: branch file/sorceryx3 r386186 - in /team/file/sorceryx3: include/asteri...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Sat Apr 20 12:40:17 CDT 2013
Author: file
Date: Sat Apr 20 12:40:14 2013
New Revision: 386186
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=386186
Log:
Add the ability to add observers to specific object types for create, update, and delete operations.
Observers receive notification of these operations in a serialized fashion using a thread pool.
This reduces the impact to callers in case the observers take a substantial amount of time to
process notifications.
Modified:
team/file/sorceryx3/include/asterisk/sorcery.h
team/file/sorceryx3/main/sorcery.c
team/file/sorceryx3/tests/test_sorcery.c
Modified: team/file/sorceryx3/include/asterisk/sorcery.h
URL: http://svnview.digium.com/svn/asterisk/team/file/sorceryx3/include/asterisk/sorcery.h?view=diff&rev=386186&r1=386185&r2=386186
==============================================================================
--- team/file/sorceryx3/include/asterisk/sorcery.h (original)
+++ team/file/sorceryx3/include/asterisk/sorcery.h Sat Apr 20 12:40:14 2013
@@ -227,6 +227,18 @@
void (*close)(void *data);
};
+/*! \brief Interface for a sorcery object type observer */
+struct ast_sorcery_observer {
+ /*! \brief Callback for when an object is created */
+ void (*created)(const void *object);
+
+ /*! \brief Callback for when an object is updated */
+ void (*updated)(const void *object);
+
+ /*! \brief Callback for when an object is deleted */
+ void (*deleted)(const void *object);
+};
+
/*! \brief Structure which contains details about a sorcery object */
struct ast_sorcery_object_details {
/*! \brief Unique identifier of this object */
@@ -536,6 +548,30 @@
int ast_sorcery_diff(const struct ast_sorcery *sorcery, const void *original, const void *modified, struct ast_variable **changes);
/*!
+ * \brief Add an observer to a specific object type
+ *
+ * \param sorcery Pointer to a sorcery structure
+ * \param type Type of object that should be observed
+ * \param callbacks Implementation of the observer interface
+ *
+ * \retval 0 success
+ * \retval -1 failure
+ */
+int ast_sorcery_observer_add(const struct ast_sorcery *sorcery, const char *type, const struct ast_sorcery_observer *callbacks);
+
+/*!
+ * \brief Remove an observer from a specific object type
+ *
+ * \param sorcery Pointer to a sorcery structure
+ * \param type Type of object that should no longer be observed
+ * \param callbacks Implementation of the observer interface
+ *
+ * \retval 0 success
+ * \retval -1 failure
+ */
+void ast_sorcery_observer_remove(const struct ast_sorcery *sorcery, const char *type, struct ast_sorcery_observer *callbacks);
+
+/*!
* \brief Create and potentially persist an object using an available wizard
*
* \param sorcery Pointer to a sorcery structure
Modified: team/file/sorceryx3/main/sorcery.c
URL: http://svnview.digium.com/svn/asterisk/team/file/sorceryx3/main/sorcery.c?view=diff&rev=386186&r1=386185&r2=386186
==============================================================================
--- team/file/sorceryx3/main/sorcery.c (original)
+++ team/file/sorceryx3/main/sorcery.c Sat Apr 20 12:40:14 2013
@@ -38,6 +38,8 @@
#include "asterisk/config_options.h"
#include "asterisk/netsock2.h"
#include "asterisk/module.h"
+#include "asterisk/taskprocessor.h"
+#include "asterisk/threadpool.h"
/* To prevent DEBUG_FD_LEAKS from interfering with things we undef open and close */
#undef open
@@ -51,6 +53,9 @@
/*! \brief Maximum length of an object field name */
#define MAX_OBJECT_FIELD 128
+
+/*! \brief Thread pool for observers */
+static struct ast_threadpool *threadpool;
/*! \brief Structure for registered object type */
struct ast_sorcery_object_type {
@@ -83,6 +88,27 @@
/*! \brief Type details */
struct aco_type type;
+
+ /*! \brief Observers */
+ struct ao2_container *observers;
+
+ /*! \brief Serializer for observers */
+ struct ast_taskprocessor *serializer;
+};
+
+/*! \brief Structure for registered object type observer */
+struct ast_sorcery_object_type_observer {
+ /*! \brief Pointer to the observer implementation */
+ const struct ast_sorcery_observer *callbacks;
+};
+
+/*! \brief Structure used for observer invocations */
+struct sorcery_observer_invocation {
+ /*! \brief Pointer to the observers */
+ struct ao2_container *observers;
+
+ /*! \brief Pointer to the object */
+ void *object;
};
/*! \brief Structure for registered object field */
@@ -213,9 +239,21 @@
int ast_sorcery_init(void)
{
+ struct ast_threadpool_options options = {
+ .version = AST_THREADPOOL_OPTIONS_VERSION,
+ .auto_increment = 1,
+ .max_size = 0,
+ .idle_timeout = 60,
+ .initial_size = 0,
+ };
ast_assert(wizards == NULL);
+ if (!(threadpool = ast_threadpool_create("Sorcery", NULL, &options))) {
+ return -1;
+ }
+
if (!(wizards = ao2_container_alloc(WIZARD_BUCKETS, sorcery_wizard_hash, sorcery_wizard_cmp))) {
+ ast_threadpool_shutdown(threadpool);
return -1;
}
@@ -317,6 +355,7 @@
ao2_cleanup(object_type->wizards);
ao2_cleanup(object_type->fields);
+ ao2_cleanup(object_type->observers);
if (object_type->info) {
aco_info_destroy(object_type->info);
@@ -324,12 +363,15 @@
}
ast_free(object_type->file);
+
+ ast_taskprocessor_unreference(object_type->serializer);
}
/*! \brief Internal function which allocates an object type structure */
static struct ast_sorcery_object_type *sorcery_object_type_alloc(const char *type, const char *module)
{
struct ast_sorcery_object_type *object_type;
+ char uuid[AST_UUID_STR_LEN];
if (!(object_type = ao2_alloc(sizeof(*object_type), sorcery_object_type_destructor))) {
return NULL;
@@ -346,12 +388,27 @@
return NULL;
}
+ if (!(object_type->observers = ao2_container_alloc_options(AO2_ALLOC_OPT_LOCK_RWLOCK, 1, NULL, NULL))) {
+ ao2_ref(object_type, -1);
+ return NULL;
+ }
+
if (!(object_type->info = ast_calloc(1, sizeof(*object_type->info) + 2 * sizeof(object_type->info->files[0])))) {
ao2_ref(object_type, -1);
return NULL;
}
if (!(object_type->file = ast_calloc(1, sizeof(*object_type->file) + 2 * sizeof(object_type->file->types[0])))) {
+ ao2_ref(object_type, -1);
+ return NULL;
+ }
+
+ if (!ast_uuid_generate_str(uuid, sizeof(uuid))) {
+ ao2_ref(object_type, -1);
+ return NULL;
+ }
+
+ if (!(object_type->serializer = ast_threadpool_serializer(uuid, threadpool))) {
ao2_ref(object_type, -1);
return NULL;
}
@@ -1015,6 +1072,55 @@
return (!object_wizard->caching && !object_wizard->wizard->create(details->sorcery, object_wizard->data, details->obj)) ? CMP_MATCH | CMP_STOP : 0;
}
+/*! \brief Destructor for observer invocation */
+static void sorcery_observer_invocation_destroy(void *obj)
+{
+ struct sorcery_observer_invocation *invocation = obj;
+
+ ao2_cleanup(invocation->observers);
+ ao2_cleanup(invocation->object);
+}
+
+/*! \brief Allocator function for observer invocation */
+static struct sorcery_observer_invocation *sorcery_observer_invocation_alloc(struct ao2_container *observers, void *object)
+{
+ struct sorcery_observer_invocation *invocation = ao2_alloc(sizeof(*invocation), sorcery_observer_invocation_destroy);
+
+ if (!invocation) {
+ return NULL;
+ }
+
+ ao2_ref(observers, +1);
+ invocation->observers = observers;
+ ao2_ref(object, +1);
+ invocation->object = object;
+
+ return invocation;
+}
+
+/*! \brief Internal callback function which notifies an individual observer that an object has been created */
+static int sorcery_observer_notify_create(void *obj, void *arg, int flags)
+{
+ const struct ast_sorcery_object_type_observer *observer = obj;
+
+ if (observer->callbacks->created) {
+ observer->callbacks->created(arg);
+ }
+
+ return 0;
+}
+
+/*! \brief Internal callback function which notifies observers that an object has been created */
+static int sorcery_observers_notify_create(void *data)
+{
+ struct sorcery_observer_invocation *invocation = data;
+
+ ao2_callback(invocation->observers, OBJ_NODATA | OBJ_MULTIPLE, sorcery_observer_notify_create, invocation->object);
+ ao2_cleanup(invocation);
+
+ return 0;
+}
+
int ast_sorcery_create(const struct ast_sorcery *sorcery, void *object)
{
const struct ast_sorcery_object_details *details = object;
@@ -1029,9 +1135,38 @@
return -1;
}
- object_wizard = ao2_callback(object_type->wizards, 0, sorcery_wizard_create, &sdetails);
+ if ((object_wizard = ao2_callback(object_type->wizards, 0, sorcery_wizard_create, &sdetails))) {
+ struct sorcery_observer_invocation *invocation = sorcery_observer_invocation_alloc(object_type->observers, object);
+
+ if (invocation && ast_taskprocessor_push(object_type->serializer, sorcery_observers_notify_create, invocation)) {
+ ao2_cleanup(invocation);
+ }
+ }
return object_wizard ? 0 : -1;
+}
+
+/*! \brief Internal callback function which notifies an individual observer that an object has been updated */
+static int sorcery_observer_notify_update(void *obj, void *arg, int flags)
+{
+ const struct ast_sorcery_object_type_observer *observer = obj;
+
+ if (observer->callbacks->updated) {
+ observer->callbacks->updated(arg);
+ }
+
+ return 0;
+}
+
+/*! \brief Internal callback function which notifies observers that an object has been updated */
+static int sorcery_observers_notify_update(void *data)
+{
+ struct sorcery_observer_invocation *invocation = data;
+
+ ao2_callback(invocation->observers, OBJ_NODATA | OBJ_MULTIPLE, sorcery_observer_notify_update, invocation->object);
+ ao2_cleanup(invocation);
+
+ return 0;
}
/*! \brief Internal function which returns if a wizard has updated the object */
@@ -1058,9 +1193,38 @@
return -1;
}
- object_wizard = ao2_callback(object_type->wizards, 0, sorcery_wizard_update, &sdetails);
+ if ((object_wizard = ao2_callback(object_type->wizards, 0, sorcery_wizard_update, &sdetails))) {
+ struct sorcery_observer_invocation *invocation = sorcery_observer_invocation_alloc(object_type->observers, object);
+
+ if (invocation && ast_taskprocessor_push(object_type->serializer, sorcery_observers_notify_update, invocation)) {
+ ao2_cleanup(invocation);
+ }
+ }
return object_wizard ? 0 : -1;
+}
+
+/*! \brief Internal callback function which notifies an individual observer that an object has been deleted */
+static int sorcery_observer_notify_delete(void *obj, void *arg, int flags)
+{
+ const struct ast_sorcery_object_type_observer *observer = obj;
+
+ if (observer->callbacks->deleted) {
+ observer->callbacks->deleted(arg);
+ }
+
+ return 0;
+}
+
+/*! \brief Internal callback function which notifies observers that an object has been deleted */
+static int sorcery_observers_notify_delete(void *data)
+{
+ struct sorcery_observer_invocation *invocation = data;
+
+ ao2_callback(invocation->observers, OBJ_NODATA | OBJ_MULTIPLE, sorcery_observer_notify_delete, invocation->object);
+ ao2_cleanup(invocation);
+
+ return 0;
}
/*! \brief Internal function which returns if a wizard has deleted the object */
@@ -1087,7 +1251,13 @@
return -1;
}
- object_wizard = ao2_callback(object_type->wizards, 0, sorcery_wizard_delete, &sdetails);
+ if ((object_wizard = ao2_callback(object_type->wizards, 0, sorcery_wizard_delete, &sdetails))) {
+ struct sorcery_observer_invocation *invocation = sorcery_observer_invocation_alloc(object_type->observers, object);
+
+ if (invocation && ast_taskprocessor_push(object_type->serializer, sorcery_observers_notify_delete, invocation)) {
+ ao2_cleanup(invocation);
+ }
+ }
return object_wizard ? 0 : -1;
}
@@ -1108,3 +1278,42 @@
const struct ast_sorcery_object_details *details = object;
return details->type;
}
+
+int ast_sorcery_observer_add(const struct ast_sorcery *sorcery, const char *type, const struct ast_sorcery_observer *callbacks)
+{
+ RAII_VAR(struct ast_sorcery_object_type *, object_type, ao2_find(sorcery->types, type, OBJ_KEY), ao2_cleanup);
+ struct ast_sorcery_object_type_observer *observer;
+
+ if (!object_type || !callbacks) {
+ return -1;
+ }
+
+ if (!(observer = ao2_alloc(sizeof(*observer), NULL))) {
+ return -1;
+ }
+
+ observer->callbacks = callbacks;
+ ao2_link(object_type->observers, observer);
+ ao2_ref(observer, -1);
+
+ return 0;
+}
+
+/*! \brief Internal callback function for removing an observer */
+static int sorcery_observer_remove(void *obj, void *arg, int flags)
+{
+ const struct ast_sorcery_object_type_observer *observer = obj;
+
+ return (observer->callbacks == arg) ? CMP_MATCH | CMP_STOP : 0;
+}
+
+void ast_sorcery_observer_remove(const struct ast_sorcery *sorcery, const char *type, struct ast_sorcery_observer *callbacks)
+{
+ RAII_VAR(struct ast_sorcery_object_type *, object_type, ao2_find(sorcery->types, type, OBJ_KEY), ao2_cleanup);
+
+ if (!object_type) {
+ return;
+ }
+
+ ao2_callback(object_type->observers, OBJ_NODATA | OBJ_UNLINK, sorcery_observer_remove, callbacks);
+}
Modified: team/file/sorceryx3/tests/test_sorcery.c
URL: http://svnview.digium.com/svn/asterisk/team/file/sorceryx3/tests/test_sorcery.c?view=diff&rev=386186&r1=386185&r2=386186
==============================================================================
--- team/file/sorceryx3/tests/test_sorcery.c (original)
+++ team/file/sorceryx3/tests/test_sorcery.c Sat Apr 20 12:40:14 2013
@@ -126,6 +126,24 @@
struct test_sorcery_object object;
};
+/*! \brief Test structure for observer */
+struct sorcery_test_observer {
+ /*! \brief Lock for notification */
+ ast_mutex_t lock;
+
+ /*! \brief Condition for notification */
+ ast_cond_t cond;
+
+ /*! \brief Pointer to the created object */
+ const void *created;
+
+ /*! \brief Pointer to the update object */
+ const void *updated;
+
+ /*! \brief Pointer to the deleted object */
+ const void *deleted;
+};
+
/*! \brief Global scope apply handler integer to make sure it executed */
static int apply_handler_called;
@@ -137,6 +155,9 @@
/*! \brief Global scope caching structure for testing */
static struct sorcery_test_caching cache = { 0, };
+
+/*! \brief Global scope observer structure for testing */
+static struct sorcery_test_observer observer;
static int sorcery_test_create(const struct ast_sorcery *sorcery, void *data, void *object)
{
@@ -170,6 +191,34 @@
.retrieve_id = sorcery_test_retrieve_id,
.update = sorcery_test_update,
.delete = sorcery_test_delete,
+};
+
+static void sorcery_observer_created(const void *object)
+{
+ SCOPED_MUTEX(lock, &observer.lock);
+ observer.created = object;
+ ast_cond_signal(&observer.cond);
+}
+
+static void sorcery_observer_updated(const void *object)
+{
+ SCOPED_MUTEX(lock, &observer.lock);
+ observer.updated = object;
+ ast_cond_signal(&observer.cond);
+}
+
+static void sorcery_observer_deleted(const void *object)
+{
+ SCOPED_MUTEX(lock, &observer.lock);
+ observer.deleted = object;
+ ast_cond_signal(&observer.cond);
+}
+
+/*! \brief Test sorcery observer implementation */
+static struct ast_sorcery_observer test_observer = {
+ .created = sorcery_observer_created,
+ .updated = sorcery_observer_updated,
+ .deleted = sorcery_observer_deleted,
};
static struct ast_sorcery *alloc_and_initialize_sorcery(void)
@@ -1945,6 +1994,131 @@
ast_test_status_update(test, "Failed to unregister test sorcery wizard\n");
return AST_TEST_FAIL;
}
+
+ return res;
+}
+
+AST_TEST_DEFINE(object_type_observer)
+{
+ RAII_VAR(struct ast_sorcery *, sorcery, NULL, ast_sorcery_unref);
+ RAII_VAR(struct test_sorcery_object *, obj, NULL, ao2_cleanup);
+ int res = AST_TEST_FAIL;
+
+ switch (cmd) {
+ case TEST_INIT:
+ info->name = "object_type_observer";
+ info->category = "/main/sorcery/";
+ info->summary = "sorcery object type observer unit test";
+ info->description =
+ "Test that object type observers get called when they should";
+ return AST_TEST_NOT_RUN;
+ case TEST_EXECUTE:
+ break;
+ }
+
+ if (!(sorcery = alloc_and_initialize_sorcery())) {
+ ast_test_status_update(test, "Failed to open sorcery structure\n");
+ return AST_TEST_FAIL;
+ }
+
+ if (!ast_sorcery_observer_add(sorcery, "test", NULL)) {
+ ast_test_status_update(test, "Successfully added a NULL observer when it should not be possible\n");
+ return AST_TEST_FAIL;
+ }
+
+ if (ast_sorcery_observer_add(sorcery, "test", &test_observer)) {
+ ast_test_status_update(test, "Failed to add a proper observer\n");
+ return AST_TEST_FAIL;
+ }
+
+ if (!(obj = ast_sorcery_alloc(sorcery, "test", "blah"))) {
+ ast_test_status_update(test, "Failed to allocate a known object type\n");
+ goto end;
+ }
+
+ ast_mutex_init(&observer.lock);
+ ast_cond_init(&observer.cond, NULL);
+ observer.created = NULL;
+ observer.updated = NULL;
+ observer.deleted = NULL;
+
+ if (ast_sorcery_create(sorcery, obj)) {
+ ast_test_status_update(test, "Failed to create object using in-memory wizard\n");
+ goto end;
+ }
+
+ ast_mutex_lock(&observer.lock);
+ while (!observer.created) {
+ struct timeval start = ast_tvnow();
+ struct timespec end = {
+ .tv_sec = start.tv_sec + 10,
+ .tv_nsec = start.tv_usec * 1000,
+ };
+ if (ast_cond_timedwait(&observer.cond, &observer.lock, &end) == ETIMEDOUT) {
+ break;
+ }
+ }
+ ast_mutex_unlock(&observer.lock);
+
+ if (!observer.created) {
+ ast_test_status_update(test, "Failed to receive observer notification for object creation within suitable timeframe\n");
+ goto end;
+ }
+
+ if (ast_sorcery_update(sorcery, obj)) {
+ ast_test_status_update(test, "Failed to update object using in-memory wizard\n");
+ goto end;
+ }
+
+ ast_mutex_lock(&observer.lock);
+ while (!observer.updated) {
+ struct timeval start = ast_tvnow();
+ struct timespec end = {
+ .tv_sec = start.tv_sec + 10,
+ .tv_nsec = start.tv_usec * 1000,
+ };
+ if (ast_cond_timedwait(&observer.cond, &observer.lock, &end) == ETIMEDOUT) {
+ break;
+ }
+ }
+ ast_mutex_unlock(&observer.lock);
+
+ if (!observer.updated) {
+ ast_test_status_update(test, "Failed to receive observer notification for object updating within suitable timeframe\n");
+ goto end;
+ }
+
+ if (ast_sorcery_delete(sorcery, obj)) {
+ ast_test_status_update(test, "Failed to delete object using in-memory wizard\n");
+ goto end;
+ }
+
+ ast_mutex_lock(&observer.lock);
+ while (!observer.deleted) {
+ struct timeval start = ast_tvnow();
+ struct timespec end = {
+ .tv_sec = start.tv_sec + 10,
+ .tv_nsec = start.tv_usec * 1000,
+ };
+ if (ast_cond_timedwait(&observer.cond, &observer.lock, &end) == ETIMEDOUT) {
+ break;
+ }
+ }
+ ast_mutex_unlock(&observer.lock);
+
+ if (!observer.deleted) {
+ ast_test_status_update(test, "Failed to receive observer notification for object deletion within suitable timeframe\n");
+ goto end;
+ }
+
+ res = AST_TEST_PASS;
+
+end:
+ observer.created = NULL;
+ observer.updated = NULL;
+ observer.deleted = NULL;
+ ast_mutex_destroy(&observer.lock);
+ ast_cond_destroy(&observer.cond);
return res;
}
@@ -2352,6 +2526,7 @@
AST_TEST_UNREGISTER(object_delete);
AST_TEST_UNREGISTER(object_delete_uncreated);
AST_TEST_UNREGISTER(caching_wizard_behavior);
+ AST_TEST_UNREGISTER(object_type_observer);
AST_TEST_UNREGISTER(configuration_file_wizard);
AST_TEST_UNREGISTER(configuration_file_wizard_with_file_integrity);
AST_TEST_UNREGISTER(configuration_file_wizard_with_criteria);
@@ -2397,6 +2572,7 @@
AST_TEST_REGISTER(object_delete);
AST_TEST_REGISTER(object_delete_uncreated);
AST_TEST_REGISTER(caching_wizard_behavior);
+ AST_TEST_REGISTER(object_type_observer);
AST_TEST_REGISTER(configuration_file_wizard);
AST_TEST_REGISTER(configuration_file_wizard_with_file_integrity);
AST_TEST_REGISTER(configuration_file_wizard_with_criteria);
More information about the asterisk-commits
mailing list