Author: andrew
Date: Fri Nov 14 12:55:20 2008
New Revision: 10403
Log:
Ton of stuff, mainly laying groundwork for bind_search functionality
Modified:
freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c
Modified: freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c
==============================================================================
--- freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c (original)
+++ freeswitch/trunk/src/mod/event_handlers/mod_erlang_event/mod_erlang_event.c Fri Nov 14 12:55:20 2008
@@ -53,14 +53,6 @@
LFLAG_STATEFUL = (1 << 8)
} event_flag_t;
-
-/* TODO - support multiple event handlers per erlang connection each with their own event filters? */
-struct event_handler {
- erlang_pid pid;
- switch_hash_t *event_hash;
- struct event_handler *next;
-};
-
struct listener {
int sockfd;
struct ei_cnode_s *ec;
@@ -70,6 +62,7 @@
switch_queue_t *log_queue;
switch_memory_pool_t *pool;
switch_mutex_t *flag_mutex;
+ switch_mutex_t *sock_mutex;
char *ebuf;
uint32_t flags;
switch_log_level_t level;
@@ -98,6 +91,19 @@
#define MAX_ACL 100
+struct erlang_binding {
+ switch_xml_section_t section;
+ erlang_pid pid;
+ char *registered_process; /* TODO */
+ listener_t *listener;
+ struct erlang_binding *next;
+};
+
+static struct {
+ struct erlang_binding *head;
+ switch_xml_binding_t *search_binding;
+} bindings;
+
static struct {
switch_mutex_t *mutex;
char *ip;
@@ -163,6 +169,49 @@
}
+/* Stolen from code added to ei in R12B-5.
+ * Since not everyone has this verison yet;
+ * provide our own version.
+ * */
+
+#define put8(s,n) do { \
+ (s)[0] = (char)((n) & 0xff); \
+ (s) += 1; \
+} while (0)
+
+#define put32be(s,n) do { \
+ (s)[0] = ((n) >> 24) & 0xff; \
+ (s)[1] = ((n) >> 16) & 0xff; \
+ (s)[2] = ((n) >> 8) & 0xff; \
+ (s)[3] = (n) & 0xff; \
+ (s) += 4; \
+} while (0)
+
+static void ei_link(listener_t *listener, erlang_pid *from, erlang_pid *to) {
+ char msgbuf[2048];
+ char *s;
+ int index = 0;
+ /*int n;*/
+
+ index = 5; /* max sizes: */
+ ei_encode_version(msgbuf,&index); /* 1 */
+ ei_encode_tuple_header(msgbuf,&index,3);
+ ei_encode_long(msgbuf,&index,ERL_LINK);
+ ei_encode_pid(msgbuf,&index,from); /* 268 */
+ ei_encode_pid(msgbuf,&index,to); /* 268 */
+
+ /* 5 byte header missing */
+ s = msgbuf;
+ put32be(s, index - 4); /* 4 */
+ put8(s, ERL_PASS_THROUGH); /* 1 */
+ /* sum: 542 */
+
+ switch_mutex_lock(listener->sock_mutex);
+ write(listener->sockfd, msgbuf, index);
+ switch_mutex_unlock(listener->sock_mutex);
+}
+
+
static void expire_listener(listener_t **listener)
{
void *pop;
@@ -180,6 +229,38 @@
}
+static void remove_binding(listener_t *listener) {
+ struct erlang_binding *ptr, *lst = NULL;
+
+ switch_mutex_lock(globals.listener_mutex);
+
+ switch_xml_set_binding_sections(bindings.search_binding, (1 << sizeof(switch_xml_section_enum_t)));
+
+ for (ptr = bindings.head; ptr; lst = ptr, ptr = ptr->next) {
+ if (ptr->listener == listener) {
+ if (bindings.head == ptr) {
+ if (ptr->next) {
+ bindings.head = ptr->next;
+ } else {
+ bindings.head = NULL;
+ break;
+ }
+ } else {
+ if (ptr->next) {
+ lst->next = ptr->next;
+ } else {
+ lst->next = NULL;
+ }
+ }
+ } else {
+ switch_xml_set_binding_sections(bindings.search_binding, switch_xml_get_binding_sections(bindings.search_binding) | ptr->section);
+ }
+ }
+
+ switch_mutex_unlock(globals.listener_mutex);
+}
+
+
static void ei_encode_switch_event(ei_x_buff *ebuf, switch_event_t *event)
{
int i;
@@ -295,40 +376,6 @@
}
-SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_erlang_event_shutdown)
-{
- listener_t *l;
- int sanity = 0;
-
- prefs.done = 1;
-
- switch_log_unbind_logger(socket_logger);
-
- /*close_socket(&listen_list.sockfd);*/
-
- while (prefs.threads || prefs.done == 1) {
- switch_yield(10000);
- if (++sanity == 1000) {
- break;
- }
- }
-
- switch_event_unbind(&globals.node);
-
- switch_mutex_lock(globals.listener_mutex);
-
- for (l = listen_list.listeners; l; l = l->next) {
- close_socket(&l->sockfd);
- }
-
- switch_mutex_unlock(globals.listener_mutex);
-
- switch_sleep(1500000); /* sleep for 1.5 seconds */
-
- return SWITCH_STATUS_SUCCESS;
-}
-
-
static void add_listener(listener_t *listener)
{
/* add me to the listeners so I get events */
@@ -357,25 +404,6 @@
switch_mutex_unlock(globals.listener_mutex);
}
-SWITCH_MODULE_LOAD_FUNCTION(mod_erlang_event_load)
-{
- switch_mutex_init(&globals.listener_mutex, SWITCH_MUTEX_NESTED, pool);
-
- if (switch_event_bind_removable(modname, SWITCH_EVENT_ALL, SWITCH_EVENT_SUBCLASS_ANY, event_handler, NULL, &globals.node) != SWITCH_STATUS_SUCCESS) {
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't bind!\n");
- close_socket(&listen_list.sockfd);
- return SWITCH_STATUS_GENERR;
- }
-
- switch_log_bind_logger(socket_logger, SWITCH_LOG_DEBUG, SWITCH_FALSE);
-
- /* connect my internal structure to the blank pointer passed to me */
- *module_interface = switch_loadable_module_create_module_interface(pool, modname);
-
- /* indicate that the module should continue to be loaded */
- return SWITCH_STATUS_SUCCESS;
-}
-
struct api_command_struct {
char *api_cmd;
@@ -456,7 +484,9 @@
ei_x_encode_string(&ebuf, acs->uuid_str);
ei_x_encode_string(&ebuf, reply);
+ switch_mutex_lock(acs->listener->sock_mutex);
ei_send(acs->listener->sockfd, &acs->pid, ebuf.buff, ebuf.index);
+ switch_mutex_unlock(acs->listener->sock_mutex);
ei_x_free(&ebuf);
}
@@ -478,7 +508,11 @@
ei_x_encode_string(&rbuf, reply);
+
+ switch_mutex_lock(acs->listener->sock_mutex);
ei_send(acs->listener->sockfd, &acs->pid, rbuf.buff, rbuf.index);
+ switch_mutex_unlock(acs->listener->sock_mutex);
+
ei_x_free(&rbuf);
}
@@ -500,6 +534,38 @@
}
+static switch_xml_t erlang_fetch (const char *sectionstr, const char *tag_name, const char *key_name, const char *key_value,
+ switch_event_t *params, void *user_data)
+{
+ switch_xml_t xml = NULL;
+ struct erlang_binding *ptr;
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "looking for bindings\n");
+
+ switch_xml_section_t section = switch_xml_parse_section_string((char *) sectionstr);
+
+ for (ptr = bindings.head; ptr && ptr->section != section; ptr = ptr->next); /* just get the first match */
+
+ if (!ptr) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "no binding for %s\n", sectionstr);
+ return NULL;
+ }
+
+ if (!ptr->listener) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "NULL pointer binding!\n");
+ return NULL; /* our pointer is trash */
+ }
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "binding for %s in section %s with key %s and value %s requested from node %s\n", tag_name, sectionstr, key_name, key_value, ptr->pid.node);
+
+
+ switch_mutex_lock(ptr->listener->sock_mutex);
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "It's a lock!\n");
+ switch_mutex_unlock(ptr->listener->sock_mutex);
+
+ return xml;
+}
+
static int handle_msg(listener_t *listener, erlang_msg *msg, ei_x_buff *buf, ei_x_buff *rbuf)
{
@@ -829,6 +895,57 @@
break;
}
+ } else if (!strncmp(tupletag, "bind", MAXATOMLEN)) {
+
+ /* format is (result|config|directory|dialplan|phrases) */
+ char sectionstr[MAXATOMLEN];
+
+ if (ei_decode_atom(buf->buff, &buf->index, sectionstr)) {
+ ei_x_encode_tuple_header(rbuf, 2);
+ ei_x_encode_atom(rbuf, "error");
+ ei_x_encode_atom(rbuf, "badarg");
+ break;
+ }
+
+ switch_xml_section_t section;
+
+ if (!(section = switch_xml_parse_section_string(sectionstr))) {
+ ei_x_encode_tuple_header(rbuf, 2);
+ ei_x_encode_atom(rbuf, "error");
+ ei_x_encode_atom(rbuf, "badarg");
+ break;
+ }
+
+ struct erlang_binding *binding, *ptr;
+
+ if (!(binding = switch_core_alloc(listener->pool, sizeof(*binding)))) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Memory Error\n");
+ ei_x_encode_tuple_header(rbuf, 2);
+ ei_x_encode_atom(rbuf, "error");
+ ei_x_encode_atom(rbuf, "badmem");
+ break;
+ }
+
+ binding->section = section;
+ binding->pid = msg->from;
+ binding->listener = listener;
+
+ switch_mutex_lock(globals.listener_mutex);
+
+ for (ptr = bindings.head; ptr && ptr->next; ptr = ptr->next);
+
+ if (ptr) {
+ ptr->next = binding;
+ } else {
+ bindings.head = binding;
+ }
+
+ switch_xml_set_binding_sections(bindings.search_binding, switch_xml_get_binding_sections(bindings.search_binding) | section);
+ switch_mutex_unlock(globals.listener_mutex);
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "sections %d\n", switch_xml_get_binding_sections(bindings.search_binding));
+
+ ei_link(listener, ei_self(listener->ec), &msg->from);
} else {
ei_x_encode_tuple_header(rbuf, 2);
@@ -852,10 +969,12 @@
switch_clear_flag_locked(listener, LFLAG_LOG);
}
} else if (!strncmp(atom, "register_log_handler", MAXATOMLEN)) {
+ ei_link(listener, ei_self(listener->ec), &msg->from);
listener->log_pid = msg->from;
listener->level = SWITCH_LOG_DEBUG;
switch_set_flag(listener, LFLAG_LOG);
} else if (!strncmp(atom, "register_event_handler", MAXATOMLEN)) {
+ ei_link(listener, ei_self(listener->ec), &msg->from);
listener->event_pid = msg->from;
if (!switch_test_flag(listener, LFLAG_EVENTS)) {
switch_set_flag_locked(listener, LFLAG_EVENTS);
@@ -888,6 +1007,10 @@
ei_x_encode_tuple_header(rbuf, 2);
ei_x_encode_atom(rbuf, "ok");
ei_x_encode_pid(rbuf, ei_self(listener->ec));
+ } else if (!strncmp(atom, "link", MAXATOMLEN)) {
+ /* debugging */
+ ei_link(listener, ei_self(listener->ec), &msg->from);
+ goto noreply;
} else {
ei_x_encode_tuple_header(rbuf, 2);
ei_x_encode_atom(rbuf, "error");
@@ -905,12 +1028,17 @@
break;
}
+
+ switch_mutex_lock(listener->sock_mutex);
ei_send(listener->sockfd, &msg->from, rbuf->buff, rbuf->index);
+ switch_mutex_unlock(listener->sock_mutex);
noreply:
return 0;
event_done:
+ switch_mutex_lock(listener->sock_mutex);
ei_send(listener->sockfd, &msg->from, rbuf->buff, rbuf->index);
+ switch_mutex_unlock(listener->sock_mutex);
return 1;
}
@@ -985,7 +1113,9 @@
ei_x_buff rbuf;
ei_x_new_with_version(&rbuf);
+ switch_mutex_lock(listener->sock_mutex);
status = ei_xreceive_msg_tmo(listener->sockfd, &msg, &buf, 100);
+ switch_mutex_unlock(listener->sock_mutex);
switch(status) {
case ERL_TICK :
@@ -1043,7 +1173,11 @@
ei_x_encode_tuple_header(&lbuf, 2);
ei_x_encode_atom(&lbuf, "log");
ei_x_encode_string(&lbuf, data);
+
+ switch_mutex_lock(listener->sock_mutex);
ei_send(listener->sockfd, &listener->log_pid, lbuf.buff, lbuf.index);
+ switch_mutex_unlock(listener->sock_mutex);
+
ei_x_free(&lbuf);
}
}
@@ -1060,7 +1194,9 @@
ei_encode_switch_event(&ebuf, pevent);
+ switch_mutex_lock(listener->sock_mutex);
ei_send(listener->sockfd, &listener->event_pid, ebuf.buff, ebuf.index);
+ switch_mutex_unlock(listener->sock_mutex);
ei_x_free(&ebuf);
switch_event_destroy(&pevent);
@@ -1084,6 +1220,9 @@
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Connection Closed\n");
switch_core_hash_destroy(&listener->event_hash);
+ /* remove any bindings for this connection */
+ remove_binding(listener);
+
if (listener->session) {
switch_channel_clear_flag(switch_core_session_get_channel(listener->session), CF_CONTROLLED);
switch_clear_flag_locked(listener, LFLAG_SESSION);
@@ -1093,6 +1232,7 @@
switch_core_destroy_memory_pool(&pool);
}
+
switch_mutex_lock(globals.listener_mutex);
prefs.threads--;
switch_mutex_unlock(globals.listener_mutex);
@@ -1111,6 +1251,7 @@
switch_threadattr_detach_set(thd_attr, 1);
switch_threadattr_stacksize_set(thd_attr, SWITCH_THREAD_STACKSIZE);
switch_thread_create(&thread, thd_attr, listener_run, listener, listener->pool);
+
}
@@ -1173,6 +1314,39 @@
}
+/* Module Hooks */
+
+
+SWITCH_MODULE_LOAD_FUNCTION(mod_erlang_event_load)
+{
+ switch_mutex_init(&globals.listener_mutex, SWITCH_MUTEX_NESTED, pool);
+
+ if (switch_event_bind_removable(modname, SWITCH_EVENT_ALL, SWITCH_EVENT_SUBCLASS_ANY, event_handler, NULL, &globals.node) != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't bind!\n");
+ close_socket(&listen_list.sockfd);
+ return SWITCH_STATUS_GENERR;
+ }
+
+ switch_log_bind_logger(socket_logger, SWITCH_LOG_DEBUG, SWITCH_FALSE);
+
+ memset(&bindings, 0, sizeof(bindings));
+
+ if (switch_xml_bind_search_function_ret(erlang_fetch, (1 << sizeof(switch_xml_section_enum_t)), NULL, &bindings.search_binding) != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Couldn't bind!\n");
+ close_socket(&listen_list.sockfd);
+ return SWITCH_STATUS_GENERR;
+ }
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "sections %d\n", switch_xml_get_binding_sections(bindings.search_binding));
+
+ /* connect my internal structure to the blank pointer passed to me */
+ *module_interface = switch_loadable_module_create_module_interface(pool, modname);
+
+ /* indicate that the module should continue to be loaded */
+ return SWITCH_STATUS_SUCCESS;
+}
+
+
SWITCH_MODULE_RUNTIME_FUNCTION(mod_erlang_event_runtime)
{
switch_memory_pool_t *pool = NULL, *listener_pool = NULL;
@@ -1328,6 +1502,7 @@
listener_pool = NULL;
listener->level = SWITCH_LOG_DEBUG;
switch_mutex_init(&listener->flag_mutex, SWITCH_MUTEX_NESTED, listener->pool);
+ switch_mutex_init(&listener->sock_mutex, SWITCH_MUTEX_NESTED, listener->pool);
switch_core_hash_init(&listener->event_hash, listener->pool);
launch_listener_thread(listener);
@@ -1358,6 +1533,41 @@
return SWITCH_STATUS_TERM;
}
+SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_erlang_event_shutdown)
+{
+ listener_t *l;
+ int sanity = 0;
+
+ prefs.done = 1;
+
+ switch_log_unbind_logger(socket_logger);
+
+ /*close_socket(&listen_list.sockfd);*/
+
+ while (prefs.threads || prefs.done == 1) {
+ switch_yield(10000);
+ if (++sanity == 1000) {
+ break;
+ }
+ }
+
+ switch_event_unbind(&globals.node);
+ switch_xml_unbind_search_function_ptr(erlang_fetch);
+
+ switch_mutex_lock(globals.listener_mutex);
+
+ for (l = listen_list.listeners; l; l = l->next) {
+ close_socket(&l->sockfd);
+ }
+
+ switch_mutex_unlock(globals.listener_mutex);
+
+ switch_sleep(1500000); /* sleep for 1.5 seconds */
+
+ return SWITCH_STATUS_SUCCESS;
+}
+
+
/* For Emacs:
* Local Variables:
* mode:c