From e83af319602b616c8cee5efc89386be3d5af61b5 Mon Sep 17 00:00:00 2001
From: Josh Perry <josh@6bit.com>
Date: Sat, 26 Mar 2011 12:34:22 -0600
Subject: [PATCH] Updated message creation

---
 .../mod_event_zmq/mod_event_zmq.cpp           | 62 +++++++++++++------
 1 file changed, 44 insertions(+), 18 deletions(-)

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
index 7c99345712..c8975ebe5d 100644
--- a/src/mod/event_handlers/mod_event_zmq/mod_event_zmq.cpp
+++ b/src/mod/event_handlers/mod_event_zmq/mod_event_zmq.cpp
@@ -1,6 +1,5 @@
 #include <switch.h>
 #include <zmq.hpp>
-#include <string>
 #include <exception>
 #include <stdexcept>
 #include <memory>
@@ -15,13 +14,6 @@ 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:
@@ -29,19 +21,28 @@ public:
 		context(1),
 		event_publisher(context, ZMQ_PUB)
 	{
-		event_publisher.bind("tcp://*.5556");
+		event_publisher.bind("tcp://*:5556");
+
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Listening for clients\n");
 	}
 
 	void PublishEvent(const switch_event_t *event) {
+		// Serialize the event into a JSON string
 		char* pjson;
 		switch_event_serialize_json(const_cast<switch_event_t*>(event), &pjson);
-		std::auto_ptr<char> json(pjson);
 
-		ZmqStringMessage msg(json.get());
+		// Use the JSON string as the message body
+		zmq::message_t msg(pjson, strlen(pjson), free_message_data, NULL);
+		
+		// Send the message
 		event_publisher.send(msg);
 	}
 
 private:
+	static void free_message_data(void *data, void *hint) {
+		free (data);
+	}
+
 	zmq::context_t context;
 	zmq::socket_t event_publisher;
 };
@@ -49,25 +50,39 @@ private:
 // Handles global inititalization and teardown of the module
 class ZmqModule {
 public:
-	ZmqModule(switch_loadable_module_interface_t **module_interface, switch_memory_pool_t *pool) {
+	ZmqModule(switch_loadable_module_interface_t **module_interface, switch_memory_pool_t *pool) :
+		_running(false) {
 		// 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.");
 		}
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Subscribed to events\n");
 
 		// Create our module interface registration
 		*module_interface = switch_loadable_module_create_module_interface(pool, modname);
+
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Module loaded\n");
 	}
 
 	void Listen() {
+		if(_running)
+			return;
+
 		_publisher.reset(new ZmqEventPublisher());
+		_running = true;
+
+		while(_running) {
+			switch_yield(100000);
+		}
 	}
 
 	~ZmqModule() {
 		// Unsubscribe from the switch events
+		_running = false;
 		switch_event_unbind(&_node);
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Module shut down\n");
 	}
 
 private:
@@ -79,11 +94,14 @@ private:
 				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());
+		} catch(...) { // Exceptions must not propogate to C caller
+			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Unknown error publishing event via 0MQ\n");
 		}
 	}
 
 	switch_event_node_t *_node;
 	std::auto_ptr<ZmqEventPublisher> _publisher;
+	bool _running;
 };
 
 //*****************************//
@@ -99,8 +117,8 @@ 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());
+	} catch(...) { // Exceptions must not propogate to C caller
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error loading 0MQ module\n");
 		return SWITCH_STATUS_GENERR;
 	}
 
@@ -110,15 +128,23 @@ SWITCH_MODULE_RUNTIME_FUNCTION(runtime) {
 	try {
 		// Begin listening for clients
 		module->Listen();
-	} catch(...) { }
+	} catch(std::exception &ex) {
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error listening for clients: %s\n", ex.what());
+	} catch(...) { // Exceptions must not propogate to C caller
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Unknown error listening for clients\n");
+	}
 
 	// Tell the switch to stop calling this runtime loop
-	return SWITCH_STATUS_FALSE;
+	return SWITCH_STATUS_TERM;
 }
 
 SWITCH_MODULE_SHUTDOWN_FUNCTION(shutdown) {
-	// Free the module object
-	module.reset();
+	try {
+		// Free the module object
+		module.reset();
+	} catch(...) { // Exceptions must not propogate to C caller
+		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Error shutting down module\n");
+	}
 }
 
 }