From 4d5540671764ae7c9bc50d66501cc5ecfa5a60ab Mon Sep 17 00:00:00 2001 From: Josh Perry Date: Fri, 25 Mar 2011 18:28:53 -0600 Subject: [PATCH] Intitial mod_event_zmq code --- src/mod/event_handlers/mod_event_zmq/Makefile | 27 ++++ .../mod_event_zmq/mod_event_zmq.cpp | 124 ++++++++++++++++++ 2 files changed, 151 insertions(+) create mode 100644 src/mod/event_handlers/mod_event_zmq/Makefile create mode 100644 src/mod/event_handlers/mod_event_zmq/mod_event_zmq.cpp diff --git a/src/mod/event_handlers/mod_event_zmq/Makefile b/src/mod/event_handlers/mod_event_zmq/Makefile new file mode 100644 index 0000000000..46d03f4bea --- /dev/null +++ b/src/mod/event_handlers/mod_event_zmq/Makefile @@ -0,0 +1,27 @@ +BASE=../../../.. + +ZMQ=zeromq-2.1.3 + +ZMQ_BASEURL=http://download.zeromq.org + +ZMQ_DIR=$(switch_srcdir)/libs/$(ZMQ) +ZMQ_BUILDDIR=$(switch_builddir)/libs/$(ZMQ) +LOCAL_CFLAGS=-I$(ZMQ_DIR)/include +ZMQ_LA=$(ZMQ_BUILDDIR)/src/libzmq.la +LOCAL_LIBADD=$(ZMQ_LA) + +include $(BASE)/build/modmake.rules + +$(ZMQ_DIR): + $(GETLIB) $(ZMQ_BASEURL) $(ZMQ).tar.gz + cd $(ZMQ_DIR) && ./autogen.sh + +$(ZMQ_BUILDDIR)/Makefile: $(ZMQ_DIR) + mkdir -p $(ZMQ_DIR) + cd $(ZMQ_BUILDDIR) && $(DEFAULT_VARS) $(ZMQ_DIR)/configure $(DEFAULT_ARGS) --srcdir=$(ZMQ_DIR) + $(TOUCH_TARGET) + +$(ZMQ_LA): $(ZMQ_BUILDDIR)/Makefile + cd $(ZMQ_BUILDDIR) && $(MAKE) + $(TOUCH_TARGET) + diff --git a/src/mod/event_handlers/mod_event_zmq/mod_event_zmq.cpp b/src/mod/event_handlers/mod_event_zmq/mod_event_zmq.cpp new file mode 100644 index 0000000000..7c99345712 --- /dev/null +++ b/src/mod/event_handlers/mod_event_zmq/mod_event_zmq.cpp @@ -0,0 +1,124 @@ +#include +#include +#include +#include +#include +#include + +namespace mod_event_zmq { + +SWITCH_MODULE_LOAD_FUNCTION(load); +SWITCH_MODULE_SHUTDOWN_FUNCTION(shutdown); +SWITCH_MODULE_RUNTIME_FUNCTION(runtime); + +extern "C" { +SWITCH_MODULE_DEFINITION(mod_event_zmq, load, shutdown, runtime); +}; + +class ZmqStringMessage : public zmq::message_t { +public: + ZmqStringMessage(const std::string &msg) { + + } +}; + +// Handles publishing events out to clients +class ZmqEventPublisher { +public: + ZmqEventPublisher() : + context(1), + event_publisher(context, ZMQ_PUB) + { + event_publisher.bind("tcp://*.5556"); + } + + void PublishEvent(const switch_event_t *event) { + char* pjson; + switch_event_serialize_json(const_cast(event), &pjson); + std::auto_ptr json(pjson); + + ZmqStringMessage msg(json.get()); + event_publisher.send(msg); + } + +private: + zmq::context_t context; + zmq::socket_t event_publisher; +}; + +// Handles global inititalization and teardown of the module +class ZmqModule { +public: + ZmqModule(switch_loadable_module_interface_t **module_interface, switch_memory_pool_t *pool) { + // Subscribe to all switch events of any subclass + // Store a pointer to ourself in the user data + if (switch_event_bind_removable(modname, SWITCH_EVENT_ALL, SWITCH_EVENT_SUBCLASS_ANY, event_handler, (void*)this, &_node) + != SWITCH_STATUS_SUCCESS) { + throw std::runtime_error("Couldn't bind to switch events."); + } + + // Create our module interface registration + *module_interface = switch_loadable_module_create_module_interface(pool, modname); + } + + void Listen() { + _publisher.reset(new ZmqEventPublisher()); + } + + ~ZmqModule() { + // Unsubscribe from the switch events + switch_event_unbind(&_node); + } + +private: + // Dispatches events to the publisher + static void event_handler(switch_event_t *event) { + try { + ZmqModule *module = (ZmqModule*)event->bind_user_data; + if(module->_publisher.get()) + module->_publisher->PublishEvent(event); + } catch(std::exception ex) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Error publishing event via 0MQ: %s\n", ex.what()); + } + } + + switch_event_node_t *_node; + std::auto_ptr _publisher; +}; + +//*****************************// +// GLOBALS // +//*****************************// +std::auto_ptr module; + + +//*****************************// +// Module interface funtions // +//*****************************// +SWITCH_MODULE_LOAD_FUNCTION(load) { + try { + module.reset(new ZmqModule(module_interface, pool)); + return SWITCH_STATUS_SUCCESS; + } catch(const std::exception &ex) { + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error loading 0MQ module: %s\n", ex.what()); + return SWITCH_STATUS_GENERR; + } + +} + +SWITCH_MODULE_RUNTIME_FUNCTION(runtime) { + try { + // Begin listening for clients + module->Listen(); + } catch(...) { } + + // Tell the switch to stop calling this runtime loop + return SWITCH_STATUS_FALSE; +} + +SWITCH_MODULE_SHUTDOWN_FUNCTION(shutdown) { + // Free the module object + module.reset(); +} + +}