FS-10820 [mod_kazoo] eventstream configuration

src/mod/event_handlers/mod_kazoo/Makefile.am

@@ -4,11 +4,23 @@ MODNAME=mod_kazoo
 if HAVE_ERLANG
 
 mod_LTLIBRARIES = mod_kazoo.la
-mod_kazoo_la_SOURCES  = mod_kazoo.c kazoo_utils.c kazoo_node.c kazoo_event_stream.c kazoo_fetch_agent.c kazoo_commands.c kazoo_dptools.c
+mod_kazoo_la_SOURCES  = mod_kazoo.c kazoo_utils.c kazoo_dptools.c kazoo_tweaks.c
+mod_kazoo_la_SOURCES  += kazoo_api.c kazoo_commands.c kazoo_config.c
+mod_kazoo_la_SOURCES  += kazoo_message.c
+mod_kazoo_la_SOURCES  += kazoo_ei_config.c kazoo_ei_utils.c kazoo_event_stream.c
+mod_kazoo_la_SOURCES  += kazoo_fetch_agent.c kazoo_node.c
+
 mod_kazoo_la_CFLAGS   = $(AM_CFLAGS) @ERLANG_CFLAGS@ -D_REENTRANT
 mod_kazoo_la_LIBADD   = $(switch_builddir)/libfreeswitch.la
 mod_kazoo_la_LDFLAGS  = -avoid-version -module -no-undefined -shared @ERLANG_LDFLAGS@
 
+mod_kazoo.la: kazoo_definitions.h $(mod_kazoo_la_OBJECTS) $(mod_kazoo_la_DEPENDENCIES) $(EXTRA_mod_kazoo_la_DEPENDENCIES)
+	$(AM_V_CCLD)$(mod_kazoo_la_LINK) $(am_mod_kazoo_la_rpath) $(mod_kazoo_la_OBJECTS) $(mod_kazoo_la_LIBADD) $(LIBS)
+
+
+kazoo_definitions.h:
+	xxd -i kazoo.conf.xml > kazoo_definitions.h
+
 else
 install: error
 all: error

src/mod/event_handlers/mod_kazoo/kazoo_commands.c

@@ -31,6 +31,7 @@
  *
  */
 #include "mod_kazoo.h"
+#include <curl/curl.h>
 #include <switch_curl.h>
 
 #define UUID_SET_DESC "Set a variable"
@@ -158,6 +159,11 @@ SWITCH_STANDARD_API(uuid_setvar_multi_function) {
 	return SWITCH_STATUS_SUCCESS;
 }
 
+static size_t body_callback(char *buffer, size_t size, size_t nitems, void *userdata)
+{
+	return size * nitems;
+}
+
 static size_t header_callback(char *buffer, size_t size, size_t nitems, void *userdata)
 {
 	switch_event_t* event = (switch_event_t*)userdata;
@@ -182,6 +188,7 @@ SWITCH_STANDARD_API(kz_http_put)
 	switch_event_t *params = NULL;
 	char *url = NULL;
 	char *filename = NULL;
+	int delete = 0;
 
 	switch_curl_slist_t *headers = NULL;  /* optional linked-list of HTTP headers */
 	char *ext;  /* file extension, used for MIME type identification */
@@ -279,6 +286,8 @@ SWITCH_STANDARD_API(kz_http_put)
 	switch_curl_easy_setopt(curl_handle, CURLOPT_USERAGENT, "freeswitch-http-cache/1.0");
 	switch_curl_easy_setopt(curl_handle, CURLOPT_HEADERDATA, stream->param_event);
 	switch_curl_easy_setopt(curl_handle, CURLOPT_HEADERFUNCTION, header_callback);
+	switch_curl_easy_setopt(curl_handle, CURLOPT_WRITEFUNCTION, body_callback);
+
 	switch_curl_easy_setopt(curl_handle, CURLOPT_SSL_VERIFYPEER, 0L);
 	switch_curl_easy_perform(curl_handle);
 	switch_curl_easy_getinfo(curl_handle, CURLINFO_RESPONSE_CODE, &httpRes);
@@ -286,22 +295,24 @@ SWITCH_STANDARD_API(kz_http_put)
 
 	if (httpRes == 200 || httpRes == 201 || httpRes == 202 || httpRes == 204) {
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "%s saved to %s\n", filename, url);
-		switch_event_add_header(stream->param_event, SWITCH_STACK_BOTTOM, "API-Output", "%s saved to %s\n", filename, url);
-		stream->write_function(stream, "+OK\n");
+		switch_event_add_header(stream->param_event, SWITCH_STACK_BOTTOM, "API-Output", "%s saved to %s", filename, url);
+		stream->write_function(stream, "+OK %s saved to %s", filename, url);
+		delete = 1;
 	} else {
 		error = switch_mprintf("Received HTTP error %ld trying to save %s to %s", httpRes, filename, url);
 		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "%s\n", error);
 		switch_event_add_header(stream->param_event, SWITCH_STACK_BOTTOM, "API-Error", "%s", error);
 		switch_event_add_header(stream->param_event, SWITCH_STACK_BOTTOM, "API-HTTP-Error", "%ld", httpRes);
-		stream->write_function(stream, "-ERR ");
-		stream->write_function(stream, error);
-		stream->write_function(stream, "\n");
+		stream->write_function(stream, "-ERR %s", error);
 		status = SWITCH_STATUS_GENERR;
 	}
 
 done:
 	if (file_to_put) {
 		fclose(file_to_put);
+		if(delete && kazoo_globals.delete_file_after_put) {
+			remove(filename);
+		}
 	}
 
 	if (headers) {

src/mod/event_handlers/mod_kazoo/kazoo_config.h

+/*
+* FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
+* Copyright (C) 2005-2012, Anthony Minessale II <anthm@freeswitch.org>
+*
+* Version: MPL 1.1
+*
+* The contents of this file are subject to the Mozilla Public License Version
+* 1.1 (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+* http://www.mozilla.org/MPL/
+*
+* Software distributed under the License is distributed on an "AS IS" basis,
+* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
+* for the specific language governing rights and limitations under the
+* License.
+*
+* The Original Code is FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
+*
+* The Initial Developer of the Original Code is
+* Anthony Minessale II <anthm@freeswitch.org>
+* Portions created by the Initial Developer are Copyright (C)
+* the Initial Developer. All Rights Reserved.
+*
+* Based on mod_skel by
+* Anthony Minessale II <anthm@freeswitch.org>
+*
+* Contributor(s):
+*
+* Daniel Bryars <danb@aeriandi.com>
+* Tim Brown <tim.brown@aeriandi.com>
+* Anthony Minessale II <anthm@freeswitch.org>
+* William King <william.king@quentustech.com>
+* Mike Jerris <mike@jerris.com>
+*
+* kazoo.c -- Sends FreeSWITCH events to an AMQP broker
+*
+*/
+
+#ifndef KAZOO_CONFIG_H
+#define KAZOO_CONFIG_H
+
+#include <switch.h>
+
+
+struct kazoo_config_t {
+	  switch_hash_t *hash;
+	  switch_memory_pool_t *pool;
+};
+
+switch_status_t kazoo_config_loglevels(switch_memory_pool_t *pool, switch_xml_t cfg, kazoo_loglevels_ptr *ptr);
+switch_status_t kazoo_config_filters(switch_memory_pool_t *pool, switch_xml_t cfg, kazoo_filter_ptr *ptr);
+switch_status_t kazoo_config_fields(kazoo_config_ptr definitions, switch_memory_pool_t *pool, switch_xml_t cfg, kazoo_fields_ptr *ptr);
+
+switch_status_t kazoo_config_event_handler(kazoo_config_ptr definitions, kazoo_config_ptr root, switch_xml_t cfg, kazoo_event_profile_ptr *ptr);
+switch_status_t kazoo_config_fetch_handler(kazoo_config_ptr definitions, kazoo_config_ptr root, switch_xml_t cfg, kazoo_fetch_profile_ptr *ptr);
+kazoo_config_ptr kazoo_config_event_handlers(kazoo_config_ptr definitions, switch_xml_t cfg);
+kazoo_config_ptr kazoo_config_fetch_handlers(kazoo_config_ptr definitions, switch_xml_t cfg);
+kazoo_config_ptr kazoo_config_definitions(switch_xml_t cfg);
+void destroy_config(kazoo_config_ptr *ptr);
+
+#endif /* KAZOO_CONFIG_H */
+

src/mod/event_handlers/mod_kazoo/kazoo_dptools.c

@@ -52,6 +52,14 @@
 #define EXPORT_LONG_DESC "Export many channel variables for the channel calling the application"
 #define EXPORT_SYNTAX "[^^<delim>]<varname>=<value> <var2>=<val2>"
 
+#define PREFIX_UNSET_SHORT_DESC "clear variables by prefix"
+#define PREFIX_UNSET_LONG_DESC "clears the channel variables that start with prefix supplied"
+#define PREFIX_UNSET_SYNTAX "<prefix>"
+
+#define UUID_MULTISET_SHORT_DESC "Set many channel variables"
+#define UUID_MULTISET_LONG_DESC "Set many channel variables for a specific channel"
+#define UUID_MULTISET_SYNTAX "<uuid> [^^<delim>]<varname>=<value> <var2>=<val2>"
+
 static void base_set (switch_core_session_t *session, const char *data, switch_stack_t stack) {
 	char *var, *val = NULL;
 
@@ -99,6 +107,22 @@ static void base_set (switch_core_session_t *session, const char *data, switch_s
 	}
 }
 
+static int kz_is_exported(switch_core_session_t *session, char *varname)
+{
+	char *array[256] = {0};
+	int i, argc;
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+	const char *exports = switch_channel_get_variable(channel, SWITCH_EXPORT_VARS_VARIABLE);
+	char *arg = switch_core_session_strdup(session, exports);
+	argc = switch_split(arg, ',', array);
+	for(i=0; i < argc; i++) {
+		if(!strcasecmp(array[i], varname))
+			return 1;
+	}
+
+	return 0;
+}
+
 static void base_export (switch_core_session_t *session, const char *data, switch_stack_t stack) {
         char *var, *val = NULL;
 
@@ -121,20 +145,52 @@ static void base_export (switch_core_session_t *session, const char *data, switc
                         }
                 }
 
-                if (val) {
-                        expanded = switch_channel_expand_variables(channel, val);
-                }
+                if(!kz_is_exported(session, var)) {
+					if (val) {
+							expanded = switch_channel_expand_variables(channel, val);
+					}
 
-                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "%s EXPORT [%s]=[%s]\n", switch_channel_get_name(channel), var,
-                                                  expanded ? expanded : "UNDEF");
-		switch_channel_export_variable_var_check(channel, var, expanded, SWITCH_EXPORT_VARS_VARIABLE, SWITCH_FALSE);
+					switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "%s EXPORT [%s]=[%s]\n", switch_channel_get_name(channel), var,
+													  expanded ? expanded : "UNDEF");
+					switch_channel_export_variable_var_check(channel, var, expanded, SWITCH_EXPORT_VARS_VARIABLE, SWITCH_FALSE);
 
-                if (expanded && expanded != val) {
-                        switch_safe_free(expanded);
+					if (expanded && expanded != val) {
+							switch_safe_free(expanded);
+					}
+                } else {
+                    switch_channel_set_variable(channel, var, val);
                 }
         }
 }
 
+SWITCH_STANDARD_APP(prefix_unset_function)
+{
+	switch_channel_t *channel = switch_core_session_get_channel(session);
+	switch_event_header_t *ei = NULL;
+	switch_event_t *clear;
+	char *arg = (char *) data;
+
+	if(switch_event_create(&clear, SWITCH_EVENT_CLONE) != SWITCH_STATUS_SUCCESS) {
+		return;
+	}
+
+	for (ei = switch_channel_variable_first(channel); ei; ei = ei->next) {
+		const char *name = ei->name;
+		char *value = ei->value;
+		if (!strncasecmp(name, arg, strlen(arg))) {
+			switch_event_add_header_string(clear, SWITCH_STACK_BOTTOM, name, value);
+		}
+	}
+
+	switch_channel_variable_last(channel);
+	for (ei = clear->headers; ei; ei = ei->next) {
+		char *varname = ei->name;
+		switch_channel_set_variable(channel, varname, NULL);
+	}
+
+	switch_event_destroy(&clear);
+}
+
 SWITCH_STANDARD_APP(multiset_function) {
 	char delim = ' ';
 	char *arg = (char *) data;
@@ -145,25 +201,79 @@ SWITCH_STANDARD_APP(multiset_function) {
 		delim = *arg++;
 	}
 
-	if (arg) {
+	if(delim != '\0') {
 		switch_channel_t *channel = switch_core_session_get_channel(session);
-		char *array[256] = {0};
-		int i, argc;
+		if (arg) {
+			char *array[256] = {0};
+			int i, argc;
 
-		arg = switch_core_session_strdup(session, arg);
-		argc = switch_split(arg, delim, array);
+			arg = switch_core_session_strdup(session, arg);
+			argc = switch_split(arg, delim, array);
 
-		for(i = 0; i < argc; i++) {
-			base_set(session, array[i], SWITCH_STACK_BOTTOM);
+			for(i = 0; i < argc; i++) {
+				base_set(session, array[i], SWITCH_STACK_BOTTOM);
+			}
 		}
-
 		if (switch_event_create(&event, SWITCH_EVENT_CHANNEL_DATA) == SWITCH_STATUS_SUCCESS) {
 			switch_channel_event_set_data(channel, event);
 			switch_event_fire(&event);
 		}
+	} else {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "multiset with empty args\n");
+	}
+}
+
+SWITCH_STANDARD_APP(uuid_multiset_function) {
 
+	char delim = ' ';
+	char *arg0 = (char *) data;
+	char *arg = strchr(arg0, ' ');
+	switch_event_t *event;
+
+
+	if(arg == NULL) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "uuid_multiset with invalid args\n");
+		return;
+	}
+	*arg = '\0';
+	arg++;
+
+	if(zstr(arg0)) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "uuid_multiset with invalid uuid\n");
+		return;
+	}
+
+
+	if (!zstr(arg) && *arg == '^' && *(arg+1) == '^') {
+		arg += 2;
+		delim = *arg++;
+	}
+
+	if(delim != '\0') {
+		switch_core_session_t *uuid_session = NULL;
+		if ((uuid_session = switch_core_session_force_locate(arg0)) != NULL) {
+			switch_channel_t *uuid_channel = switch_core_session_get_channel(uuid_session);
+			if (arg) {
+				char *array[256] = {0};
+				int i, argc;
+
+				arg = switch_core_session_strdup(session, arg);
+				argc = switch_split(arg, delim, array);
+
+				for(i = 0; i < argc; i++) {
+					base_set(uuid_session, array[i], SWITCH_STACK_BOTTOM);
+				}
+			}
+			if (switch_event_create(&event, SWITCH_EVENT_CHANNEL_DATA) == SWITCH_STATUS_SUCCESS) {
+				switch_channel_event_set_data(uuid_channel, event);
+				switch_event_fire(&event);
+			}
+			switch_core_session_rwunlock(uuid_session);
+		} else {
+			base_set(session, data, SWITCH_STACK_BOTTOM);
+		}
 	} else {
-		base_set(session, data, SWITCH_STACK_BOTTOM);
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "multiset with empty args\n");
 	}
 }
 
@@ -205,19 +315,23 @@ SWITCH_STANDARD_APP(multiunset_function) {
 		delim = *arg++;
 	}
 
-	if (arg) {
-		char *array[256] = {0};
-		int i, argc;
+	if(delim != '\0') {
+		if (arg) {
+			char *array[256] = {0};
+			int i, argc;
 
-		arg = switch_core_session_strdup(session, arg);
-		argc = switch_split(arg, delim, array);
+			arg = switch_core_session_strdup(session, arg);
+			argc = switch_split(arg, delim, array);
 
-		for(i = 0; i < argc; i++) {
-			switch_channel_set_variable(switch_core_session_get_channel(session), array[i], NULL);
-		}
+			for(i = 0; i < argc; i++) {
+				switch_channel_set_variable(switch_core_session_get_channel(session), array[i], NULL);
+			}
 
+		} else {
+			switch_channel_set_variable(switch_core_session_get_channel(session), arg, NULL);
+		}
 	} else {
-		switch_channel_set_variable(switch_core_session_get_channel(session), arg, NULL);
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "multiunset with empty args\n");
 	}
 }
 
@@ -230,18 +344,22 @@ SWITCH_STANDARD_APP(export_function) {
                 delim = *arg++;
         }
 
-        if (arg) {
-                char *array[256] = {0};
-                int i, argc;
+        if(delim != '\0') {
+			if (arg) {
+					char *array[256] = {0};
+					int i, argc;
 
-                arg = switch_core_session_strdup(session, arg);
-                argc = switch_split(arg, delim, array);
+					arg = switch_core_session_strdup(session, arg);
+					argc = switch_split(arg, delim, array);
 
-                for(i = 0; i < argc; i++) {
-                        base_export(session, array[i], SWITCH_STACK_BOTTOM);
-                }
+					for(i = 0; i < argc; i++) {
+							base_export(session, array[i], SWITCH_STACK_BOTTOM);
+					}
+			} else {
+					base_export(session, data, SWITCH_STACK_BOTTOM);
+			}
         } else {
-                base_export(session, data, SWITCH_STACK_BOTTOM);
+    		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "export with empty args\n");
         }
 }
 
@@ -254,6 +372,11 @@ void add_kz_dptools(switch_loadable_module_interface_t **module_interface, switc
 				   SAF_SUPPORT_NOMEDIA | SAF_ROUTING_EXEC | SAF_ZOMBIE_EXEC);
 	SWITCH_ADD_APP(app_interface, "kz_multiunset", MULTISET_SHORT_DESC, MULTISET_LONG_DESC, multiunset_function, MULTIUNSET_SYNTAX,
 				   SAF_SUPPORT_NOMEDIA | SAF_ROUTING_EXEC | SAF_ZOMBIE_EXEC);
-        SWITCH_ADD_APP(app_interface, "kz_export", EXPORT_SHORT_DESC, EXPORT_LONG_DESC, export_function, EXPORT_SYNTAX,
-                                   SAF_SUPPORT_NOMEDIA | SAF_ROUTING_EXEC | SAF_ZOMBIE_EXEC);
+	SWITCH_ADD_APP(app_interface, "kz_export", EXPORT_SHORT_DESC, EXPORT_LONG_DESC, export_function, EXPORT_SYNTAX,
+	               SAF_SUPPORT_NOMEDIA | SAF_ROUTING_EXEC | SAF_ZOMBIE_EXEC);
+	SWITCH_ADD_APP(app_interface, "kz_prefix_unset", PREFIX_UNSET_SHORT_DESC, PREFIX_UNSET_LONG_DESC, prefix_unset_function, PREFIX_UNSET_SYNTAX,
+	               SAF_SUPPORT_NOMEDIA | SAF_ROUTING_EXEC | SAF_ZOMBIE_EXEC);
+	SWITCH_ADD_APP(app_interface, "kz_uuid_multiset", UUID_MULTISET_SHORT_DESC, UUID_MULTISET_LONG_DESC, uuid_multiset_function, UUID_MULTISET_SYNTAX,
+				   SAF_SUPPORT_NOMEDIA | SAF_ROUTING_EXEC | SAF_ZOMBIE_EXEC);
+
 }

src/mod/event_handlers/mod_kazoo/kazoo_ei.h

+#ifndef KAZOO_EI_H
+#define KAZOO_EI_H
+
+#include <switch.h>
+#include <ei.h>
+
+#define MODNAME "mod_kazoo"
+#define BUNDLE "community"
+#define RELEASE "v1.5.0-1"
+#define VERSION "mod_kazoo v1.5.0-1 community"
+
+typedef enum {KAZOO_FETCH_PROFILE, KAZOO_EVENT_PROFILE} kazoo_profile_type;
+
+typedef enum {ERLANG_TUPLE, ERLANG_MAP} kazoo_json_term;
+
+typedef struct ei_xml_agent_s ei_xml_agent_t;
+typedef ei_xml_agent_t *ei_xml_agent_ptr;
+
+typedef struct kazoo_event kazoo_event_t;
+typedef kazoo_event_t *kazoo_event_ptr;
+
+typedef struct kazoo_event_profile kazoo_event_profile_t;
+typedef kazoo_event_profile_t *kazoo_event_profile_ptr;
+
+typedef struct kazoo_fetch_profile kazoo_fetch_profile_t;
+typedef kazoo_fetch_profile_t *kazoo_fetch_profile_ptr;
+
+typedef struct kazoo_config_t kazoo_config;
+typedef kazoo_config *kazoo_config_ptr;
+
+#include "kazoo_fields.h"
+#include "kazoo_config.h"
+
+struct ei_send_msg_s {
+	ei_x_buff buf;
+	erlang_pid pid;
+};
+typedef struct ei_send_msg_s ei_send_msg_t;
+
+struct ei_received_msg_s {
+	ei_x_buff buf;
+	erlang_msg msg;
+};
+typedef struct ei_received_msg_s ei_received_msg_t;
+
+
+typedef struct ei_event_stream_s ei_event_stream_t;
+typedef struct ei_node_s ei_node_t;
+
+struct ei_event_binding_s {
+	char id[SWITCH_UUID_FORMATTED_LENGTH + 1];
+	switch_event_node_t *node;
+	switch_event_types_t type;
+	const char *subclass_name;
+	ei_event_stream_t* stream;
+	kazoo_event_ptr event;
+
+	struct ei_event_binding_s *next;
+};
+typedef struct ei_event_binding_s ei_event_binding_t;
+
+struct ei_event_stream_s {
+	switch_memory_pool_t *pool;
+	ei_event_binding_t *bindings;
+	switch_queue_t *queue;
+	switch_socket_t *acceptor;
+	switch_pollset_t *pollset;
+	switch_pollfd_t *pollfd;
+	switch_socket_t *socket;
+	switch_mutex_t *socket_mutex;
+	switch_bool_t connected;
+	char remote_ip[48];
+	uint16_t remote_port;
+	char local_ip[48];
+	uint16_t local_port;
+	erlang_pid pid;
+	uint32_t flags;
+	ei_node_t *node;
+	struct ei_event_stream_s *next;
+};
+
+struct ei_node_s {
+	int nodefd;
+	switch_atomic_t pending_bgapi;
+	switch_atomic_t receive_handlers;
+	switch_memory_pool_t *pool;
+	ei_event_stream_t *event_streams;
+	switch_mutex_t *event_streams_mutex;
+	switch_queue_t *send_msgs;
+	switch_queue_t *received_msgs;
+	char *peer_nodename;
+	switch_time_t created_time;
+	switch_socket_t *socket;
+	char remote_ip[48];
+	uint16_t remote_port;
+	char local_ip[48];
+	uint16_t local_port;
+	uint32_t flags;
+	int legacy;
+	struct ei_node_s *next;
+};
+
+
+struct xml_fetch_reply_s {
+	char uuid_str[SWITCH_UUID_FORMATTED_LENGTH + 1];
+	char *xml_str;
+	struct xml_fetch_reply_s *next;
+};
+typedef struct xml_fetch_reply_s xml_fetch_reply_t;
+
+struct fetch_handler_s {
+	erlang_pid pid;
+	struct fetch_handler_s *next;
+};
+typedef struct fetch_handler_s fetch_handler_t;
+
+struct ei_xml_client_s {
+	ei_node_t *ei_node;
+	fetch_handler_t *fetch_handlers;
+	struct ei_xml_client_s *next;
+};
+typedef struct ei_xml_client_s ei_xml_client_t;
+
+struct ei_xml_agent_s {
+	switch_memory_pool_t *pool;
+	switch_xml_section_t section;
+	switch_thread_rwlock_t *lock;
+	ei_xml_client_t *clients;
+	switch_mutex_t *current_client_mutex;
+	ei_xml_client_t *current_client;
+	switch_mutex_t *replies_mutex;
+	switch_thread_cond_t *new_reply;
+	xml_fetch_reply_t *replies;
+	kazoo_fetch_profile_ptr profile;
+
+};
+
+struct globals_s {
+	switch_memory_pool_t *pool;
+	switch_atomic_t threads;
+	switch_socket_t *acceptor;
+	struct ei_cnode_s ei_cnode;
+	switch_thread_rwlock_t *ei_nodes_lock;
+	ei_node_t *ei_nodes;
+
+	switch_xml_binding_t *config_fetch_binding;
+	switch_xml_binding_t *directory_fetch_binding;
+	switch_xml_binding_t *dialplan_fetch_binding;
+	switch_xml_binding_t *channels_fetch_binding;
+	switch_xml_binding_t *languages_fetch_binding;
+	switch_xml_binding_t *chatplan_fetch_binding;
+
+	switch_hash_t *event_filter;
+	int epmdfd;
+	int num_worker_threads;
+	switch_bool_t nat_map;
+	switch_bool_t ei_shortname;
+	int ei_compat_rel;
+	char *ip;
+	char *hostname;
+	char *ei_cookie;
+	char *ei_nodename;
+	char *kazoo_var_prefix;
+	int var_prefix_length;
+	uint32_t flags;
+	int send_all_headers;
+	int send_all_private_headers;
+	int connection_timeout;
+	int receive_timeout;
+	int receive_msg_preallocate;
+	int event_stream_preallocate;
+	int send_msg_batch;
+	short event_stream_framing;
+	switch_port_t port;
+	int config_fetched;
+	int io_fault_tolerance;
+	kazoo_event_profile_ptr events;
+	kazoo_config_ptr definitions;
+	kazoo_config_ptr event_handlers;
+	kazoo_config_ptr fetch_handlers;
+	kazoo_json_term  json_encoding;
+
+	int delete_file_after_put;
+	int enable_legacy;
+
+};
+typedef struct globals_s globals_t;
+extern globals_t kazoo_globals;
+
+/* kazoo_event_stream.c */
+ei_event_stream_t *find_event_stream(ei_event_stream_t *event_streams, const erlang_pid *from);
+
+//ei_event_stream_t *new_event_stream(ei_event_stream_t **event_streams, const erlang_pid *from);
+ei_event_stream_t *new_event_stream(ei_node_t *ei_node, const erlang_pid *from);
+
+
+switch_status_t remove_event_stream(ei_event_stream_t **event_streams, const erlang_pid *from);
+switch_status_t remove_event_streams(ei_event_stream_t **event_streams);
+unsigned long get_stream_port(const ei_event_stream_t *event_stream);
+switch_status_t add_event_binding(ei_event_stream_t *event_stream, const char *event_name);
+//switch_status_t add_event_binding(ei_event_stream_t *event_stream, const switch_event_types_t event_type, const char *subclass_name);
+switch_status_t remove_event_binding(ei_event_stream_t *event_stream, const switch_event_types_t event_type, const char *subclass_name);
+switch_status_t remove_event_bindings(ei_event_stream_t *event_stream);
+
+/* kazoo_node.c */
+switch_status_t new_kazoo_node(int nodefd, ErlConnect *conn);
+
+/* kazoo_ei_utils.c */
+void close_socket(switch_socket_t **sock);
+void close_socketfd(int *sockfd);
+switch_socket_t *create_socket_with_port(switch_memory_pool_t *pool, switch_port_t port);
+switch_socket_t *create_socket(switch_memory_pool_t *pool);
+switch_status_t create_ei_cnode(const char *ip_addr, const char *name, struct ei_cnode_s *ei_cnode);
+switch_status_t ei_compare_pids(const erlang_pid *pid1, const erlang_pid *pid2);
+void ei_encode_switch_event_headers(ei_x_buff *ebuf, switch_event_t *event);
+void ei_encode_switch_event_headers_2(ei_x_buff *ebuf, switch_event_t *event, int decode);
+void ei_encode_json(ei_x_buff *ebuf, cJSON *JObj);
+void ei_link(ei_node_t *ei_node, erlang_pid * from, erlang_pid * to);
+void ei_encode_switch_event(ei_x_buff * ebuf, switch_event_t *event);
+int ei_helper_send(ei_node_t *ei_node, erlang_pid* to, ei_x_buff *buf);
+int ei_decode_atom_safe(char *buf, int *index, char *dst);
+int ei_decode_string_or_binary_limited(char *buf, int *index, int maxsize, char *dst);
+int ei_decode_string_or_binary(char *buf, int *index, char **dst);
+switch_status_t create_acceptor();
+switch_hash_t *create_default_filter();
+
+void fetch_config();
+
+switch_status_t kazoo_load_config();
+void kazoo_destroy_config();
+
+
+#define _ei_x_encode_string(buf, string) { ei_x_encode_binary(buf, string, strlen(string)); }
+
+/* kazoo_fetch_agent.c */
+switch_status_t bind_fetch_agents();
+switch_status_t unbind_fetch_agents();
+switch_status_t remove_xml_clients(ei_node_t *ei_node);
+switch_status_t add_fetch_handler(ei_node_t *ei_node, erlang_pid *from, switch_xml_binding_t *binding);
+switch_status_t remove_fetch_handlers(ei_node_t *ei_node, erlang_pid *from);
+switch_status_t fetch_reply(char *uuid_str, char *xml_str, switch_xml_binding_t *binding);
+switch_status_t handle_api_command_streams(ei_node_t *ei_node, switch_stream_handle_t *stream);
+
+void bind_event_profiles(kazoo_event_ptr event);
+void rebind_fetch_profiles(kazoo_config_ptr fetch_handlers);
+switch_status_t kazoo_config_handlers(switch_xml_t cfg);
+
+/* runtime */
+SWITCH_MODULE_RUNTIME_FUNCTION(mod_kazoo_runtime);
+
+#endif /* KAZOO_EI_H */
+
+/* For Emacs:
+ * Local Variables:
+ * mode:c
+ * indent-tabs-mode:t
+ * tab-width:4
+ * c-basic-offset:4
+ * End:
+ * For VIM:
+ * vim:set softtabstop=4 shiftwidth=4 tabstop=4:
+ */

src/mod/event_handlers/mod_kazoo/kazoo_fields.h

+/*
+* FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
+* Copyright (C) 2005-2012, Anthony Minessale II <anthm@freeswitch.org>
+*
+* Version: MPL 1.1
+*
+* The contents of this file are subject to the Mozilla Public License Version
+* 1.1 (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+* http://www.mozilla.org/MPL/
+*
+* Software distributed under the License is distributed on an "AS IS" basis,
+* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
+* for the specific language governing rights and limitations under the
+* License.
+*
+* The Original Code is FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
+*
+* The Initial Developer of the Original Code is
+* Anthony Minessale II <anthm@freeswitch.org>
+* Portions created by the Initial Developer are Copyright (C)
+* the Initial Developer. All Rights Reserved.
+*
+* Based on mod_skel by
+* Anthony Minessale II <anthm@freeswitch.org>
+*
+* Contributor(s):
+*
+* Daniel Bryars <danb@aeriandi.com>
+* Tim Brown <tim.brown@aeriandi.com>
+* Anthony Minessale II <anthm@freeswitch.org>
+* William King <william.king@quentustech.com>
+* Mike Jerris <mike@jerris.com>
+*
+* kazoo.c -- Sends FreeSWITCH events to an AMQP broker
+*
+*/
+
+#ifndef KAZOO_FIELDS_H
+#define KAZOO_FIELDS_H
+
+#include <switch.h>
+
+#define MAX_LIST_FIELDS 25
+
+typedef struct kazoo_log_levels kazoo_loglevels_t;
+typedef kazoo_loglevels_t *kazoo_loglevels_ptr;
+
+struct kazoo_log_levels
+{
+	switch_log_level_t success_log_level;
+	switch_log_level_t failed_log_level;
+	switch_log_level_t warn_log_level;
+	switch_log_level_t info_log_level;
+	switch_log_level_t time_log_level;
+	switch_log_level_t filtered_event_log_level;
+	switch_log_level_t filtered_field_log_level;
+
+};
+
+typedef struct kazoo_logging kazoo_logging_t;
+typedef kazoo_logging_t *kazoo_logging_ptr;
+
+struct kazoo_logging
+{
+	kazoo_loglevels_ptr levels;
+	const char *profile_name;
+	const char *event_name;
+};
+
+typedef struct kazoo_list_s {
+  char *value[MAX_LIST_FIELDS];
+  int size;
+} kazoo_list_t;
+
+typedef enum {
+	FILTER_COMPARE_REGEX,
+	FILTER_COMPARE_LIST,
+	FILTER_COMPARE_VALUE,
+	FILTER_COMPARE_PREFIX,
+	FILTER_COMPARE_EXISTS
+} kazoo_filter_compare_type;
+
+typedef enum {
+	FILTER_EXCLUDE,
+	FILTER_INCLUDE,
+	FILTER_ENSURE
+} kazoo_filter_type;
+
+typedef struct kazoo_filter_t {
+	kazoo_filter_type type;
+	kazoo_filter_compare_type compare;
+	char* name;
+	char* value;
+	kazoo_list_t list;
+	struct kazoo_filter_t* next;
+} kazoo_filter, *kazoo_filter_ptr;
+
+
+typedef enum {
+	JSON_NONE,
+	JSON_STRING,
+	JSON_NUMBER,
+	JSON_BOOLEAN,
+	JSON_OBJECT,
+	JSON_RAW
+} kazoo_json_field_type;
+
+typedef enum {
+	FIELD_NONE,
+	FIELD_COPY,
+	FIELD_STATIC,
+	FIELD_FIRST_OF,
+	FIELD_EXPAND,
+	FIELD_PREFIX,
+	FIELD_OBJECT,
+	FIELD_GROUP,
+	FIELD_REFERENCE,
+
+} kazoo_field_type;
+
+typedef struct kazoo_field_t kazoo_field;
+typedef kazoo_field *kazoo_field_ptr;
+
+typedef struct kazoo_fields_t kazoo_fields;
+typedef kazoo_fields *kazoo_fields_ptr;
+
+typedef struct kazoo_definition_t kazoo_definition;
+typedef kazoo_definition *kazoo_definition_ptr;
+
+struct kazoo_field_t {
+	char* name;
+	char* value;
+	char* as;
+	kazoo_list_t list;
+	switch_bool_t exclude_prefix;
+	kazoo_field_type in_type;
+	kazoo_json_field_type out_type;
+	kazoo_filter_ptr filter;
+
+	kazoo_definition_ptr ref;
+	kazoo_field_ptr next;
+	kazoo_fields_ptr children;
+};
+
+struct kazoo_fields_t {
+	kazoo_field_ptr head;
+	int verbose;
+};
+
+
+struct kazoo_definition_t {
+	char* name;
+	kazoo_field_ptr head;
+	kazoo_filter_ptr filter;
+};
+
+struct kazoo_event {
+	kazoo_event_profile_ptr profile;
+	char *name;
+	kazoo_fields_ptr fields;
+	kazoo_filter_ptr filter;
+
+	kazoo_event_t* next;
+};
+
+struct kazoo_event_profile {
+	char *name;
+	kazoo_config_ptr root;
+	switch_bool_t running;
+	switch_memory_pool_t *pool;
+	kazoo_filter_ptr filter;
+	kazoo_fields_ptr fields;
+	kazoo_event_ptr events;
+
+	kazoo_loglevels_ptr logging;
+};
+
+struct kazoo_fetch_profile {
+	char *name;
+	kazoo_config_ptr root;
+	switch_bool_t running;
+	switch_memory_pool_t *pool;
+	kazoo_fields_ptr fields;
+	int fetch_timeout;
+	switch_mutex_t *fetch_reply_mutex;
+	switch_hash_t *fetch_reply_hash;
+	switch_xml_binding_t *fetch_binding;
+	switch_xml_section_t section;
+
+	kazoo_loglevels_ptr logging;
+};
+
+#endif /* KAZOO_FIELDS_H */
+

src/mod/event_handlers/mod_kazoo/kazoo_message.h

+/*
+* FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
+* Copyright (C) 2005-2012, Anthony Minessale II <anthm@freeswitch.org>
+*
+* Version: MPL 1.1
+*
+* The contents of this file are subject to the Mozilla Public License Version
+* 1.1 (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+* http://www.mozilla.org/MPL/
+*
+* Software distributed under the License is distributed on an "AS IS" basis,
+* WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
+* for the specific language governing rights and limitations under the
+* License.
+*
+* The Original Code is FreeSWITCH Modular Media Switching Software Library / Soft-Switch Application
+*
+* The Initial Developer of the Original Code is
+* Anthony Minessale II <anthm@freeswitch.org>
+* Portions created by the Initial Developer are Copyright (C)
+* the Initial Developer. All Rights Reserved.
+*
+* Based on mod_skel by
+* Anthony Minessale II <anthm@freeswitch.org>
+*
+* Contributor(s):
+*
+* Daniel Bryars <danb@aeriandi.com>
+* Tim Brown <tim.brown@aeriandi.com>
+* Anthony Minessale II <anthm@freeswitch.org>
+* William King <william.king@quentustech.com>
+* Mike Jerris <mike@jerris.com>
+*
+* kazoo.c -- Sends FreeSWITCH events to an AMQP broker
+*
+*/
+
+#ifndef KAZOO_MESSAGE_H
+#define KAZOO_MESSAGE_H
+
+#include <switch.h>
+
+typedef struct {
+    uint64_t timestamp;
+    uint64_t start;
+    uint64_t filter;
+    uint64_t serialize;
+    uint64_t print;
+    uint64_t rk;
+} kazoo_message_times_t, *kazoo_message_times_ptr;
+
+typedef struct {
+    cJSON *JObj;
+} kazoo_message_t, *kazoo_message_ptr;
+
+
+kazoo_message_ptr kazoo_message_create_event(switch_event_t* evt, kazoo_event_ptr event, kazoo_event_profile_ptr profile);
+kazoo_message_ptr kazoo_message_create_fetch(switch_event_t* evt, kazoo_fetch_profile_ptr profile);
+
+void kazoo_message_destroy(kazoo_message_ptr *msg);
+
+
+#endif /* KAZOO_MESSAGE_H */
+