Add some message types and reqresp types for read data from plugin

include/neu_adapter.h

@@ -31,8 +31,12 @@ typedef struct neu_adapter neu_adapter_t;
 
 typedef enum neu_reqresp_type {
 	NEU_REQRESP_NOP,
+	NEU_REQRESP_START_PERIODIC_READ,
+	NEU_REQRESP_STOP_PERIODIC_READ,
 	NEU_REQRESP_READ,
 	NEU_REQRESP_WRITE,
+	// The ownership of buffer will move into reqresp.
+	NEU_REQRESP_MOVE_BUF,
 } neu_reqresp_type_e;
 
 typedef struct neu_request {

plugins/plugin_sample.c

@@ -33,8 +33,8 @@ static neu_plugin_t* sample_plugin_open(neu_adapter_t* adapter,
 {
 	neu_plugin_t* plugin;
 
-	if (adapter == NULL) {
-		log_error("Open plugin with NULL adapter");
+	if (adapter == NULL || callbacks == NULL) {
+		log_error("Open plugin with NULL adapter or callbacks");
 		return NULL;
 	}
 
@@ -89,20 +89,35 @@ static int sample_plugin_request(neu_plugin_t* plugin,
 {
 	int rv = 0;
 
-	if (plugin == NULL) {
-		log_warn("The plugin pointer is NULL");
+	if (plugin == NULL || req == NULL) {
+		log_warn("The plugin pointer or request is NULL");
 		return (-1);
 	}
 
 	log_info("send request to plugin: %s", neu_plugin_module.module_name);
 	const adapter_callbacks_t* adapter_callbacks;
 	adapter_callbacks = plugin->common.adapter_callbacks;
-	if (adapter_callbacks != NULL) {
-		neu_response_t resp;
 
-		memset(&resp, 0, sizeof(resp));
-		resp.req_id = req->req_id;
-		rv = adapter_callbacks->response(plugin->common.adapter, &resp);
+	switch (req->req_type) {
+		case NEU_REQRESP_READ:
+		{
+			neu_response_t resp;
+			static const char* resp_str = "Sample plugin read response";
+
+			char* resp_buf;
+			resp_buf = strdup(resp_str);
+
+			memset(&resp, 0, sizeof(resp));
+			resp.req_id    = req->req_id;
+			resp.resp_type = NEU_REQRESP_MOVE_BUF;
+			resp.buf_len   = sizeof(strlen(resp_buf) + 1);
+			resp.buf       = resp_buf;
+			rv = adapter_callbacks->response(plugin->common.adapter, &resp);
+			break;
+		}
+
+		default:
+			break;
 	}
 	return rv;
 }

src/adapter/adapter.c

@@ -45,6 +45,7 @@ typedef enum adapter_state {
 struct neu_adapter {
 	uint32_t	    id;
 	adapter_type_e	type;
+	nng_mtx*		mtx;
 	adapter_state_e state;
 	bool			stop;
 	char*			name;
@@ -52,6 +53,7 @@ struct neu_adapter {
 	nng_pipe		pipe;
 	nng_socket		sock;
 	nng_thread*     thrd;
+	uint32_t			 new_req_id;
 	char*				 plugin_lib_name;
 	void*           	 plugin_lib;		// handle of dynamic lib
 	neu_plugin_module_t* plugin_module;
@@ -89,17 +91,20 @@ static int unload_plugin_library(void* lib_handle)
 	return rv;
 }
 
+static uint32_t adapter_get_req_id(neu_adapter_t* adapter)
+{
+	uint32_t req_id;
+
+	req_id = adapter->new_req_id++;
+	return req_id;
+}
+
 static void adapter_loop(void* arg)
 {
 	int rv;
 	neu_adapter_t* adapter;
 
 	adapter = (neu_adapter_t*)arg;
-	rv = nng_pair1_open_poly(&adapter->sock);
-	if (rv != 0) {
-		neu_panic("The adapter(%s) can't open pipe", adapter->name);
-	}
-
 	const char* manager_url;
 	manager_url = neu_manager_get_url(adapter->manager);
 	rv = nng_dial(adapter->sock, manager_url, NULL, 0);
@@ -107,15 +112,19 @@ static void adapter_loop(void* arg)
 		neu_panic("The adapter can't dial to %s", manager_url);
 	}
 
+	nng_mtx_lock(adapter->mtx);
+	adapter->state = ADAPTER_STATE_IDLE;
+	nng_mtx_unlock(adapter->mtx);
+
 	const char* adapter_str = "adapter started";
 	nng_msg* out_msg;
 	size_t msg_size;
 	msg_size = msg_inplace_data_get_size(strlen(adapter_str) + 1);
 	rv = nng_msg_alloc(&out_msg, msg_size);
 	if (rv == 0) {
-		void* msg_ptr;
+		message_t* msg_ptr;
 		char* buf_ptr;
-		msg_ptr = nng_msg_body(out_msg);
+		msg_ptr = (message_t*)nng_msg_body(out_msg);
 		msg_inplace_data_init(msg_ptr, MSG_EVENT_STATUS_STRING,
 							  msg_size);
 		buf_ptr = msg_get_buf_ptr(msg_ptr);
@@ -127,29 +136,69 @@ static void adapter_loop(void* arg)
 	while (1) {
 		nng_msg* msg;
 
+		nng_mtx_lock(adapter->mtx);
+		if (adapter->stop) {
+			adapter->state = ADAPTER_STATE_NULL;
+			nng_mtx_unlock(adapter->mtx);
+			log_info("Exit loop of the adapter(%s)", adapter->name);
+			break;
+		}
+		nng_mtx_unlock(adapter->mtx);
+
 		rv = nng_recvmsg(adapter->sock, &msg, 0);
 		if (rv != 0) {
-			if (adapter->stop) {
-				break;
-			} else {
-				log_warn("Manage pipe no message received");
-				continue;
-			}
+			log_warn("Manage pipe no message received");
+			continue;
 		}
 
-		message_t* neu_msg;
-		neu_msg = nng_msg_body(msg);
-		switch (msg_get_type(neu_msg)) {
+		message_t* pay_msg;
+		pay_msg = nng_msg_body(msg);
+		switch (msg_get_type(pay_msg)) {
 			case MSG_CONFIG_INFO_STRING:
 			{
 				char* buf_ptr;
-				buf_ptr = msg_get_buf_ptr(neu_msg);
+				buf_ptr = msg_get_buf_ptr(pay_msg);
 				log_info("Recieve string: %s", buf_ptr);
 				break;
 			}
 
+			case MSG_CMD_START_READ:
+			{
+				const neu_plugin_intf_funs_t* intf_funs;
+				neu_request_t req;
+				uint32_t      req_code;
+				intf_funs = adapter->plugin_module->intf_funs;
+				req.req_id   = adapter_get_req_id(adapter);
+				req.req_type = NEU_REQRESP_READ;
+				req.buf_len  = sizeof(uint32_t);
+				req_code     = 1;
+				req.buf      = (char*)&req_code;
+				intf_funs->request(adapter->plugin, &req);
+				break;
+			}
+
+			case MSG_CMD_STOP_READ:
+			{
+				break;
+			}
+
+			case MSG_CMD_EXIT_LOOP:
+			{
+				uint32_t exit_code;
+
+				exit_code = *(uint32_t*)msg_get_buf_ptr(pay_msg);
+				log_info("adapter(%s) exit loop by exit_code=%d",
+					     adapter->name, exit_code);
+				nng_mtx_lock(adapter->mtx);
+				adapter->state = ADAPTER_STATE_NULL;
+				adapter->stop  = true;
+				nng_mtx_unlock(adapter->mtx);
+				break;
+			}
+
 			default:
-				log_warn("Receive a not handled message(type: %d)", msg_get_type(neu_msg));
+				log_warn("Receive a not supported message(type: %d)",
+					   	 msg_get_type(pay_msg));
 				break;
 		}
 
@@ -163,11 +212,44 @@ static int adapter_response(neu_adapter_t* adapter, neu_response_t* resp)
 {
 	int rv = 0;
 
+	if (adapter == NULL || resp == NULL) {
+		log_warn("The adapter or response is NULL");
+		return (-1);
+	}
+
 	log_info("Get response from plugin");
+	switch (resp->resp_type) {
+		case NEU_REQRESP_MOVE_BUF:
+		{
+			core_databuf_t* databuf;
+			databuf = core_databuf_new_with_buf(resp->buf, resp->buf_len);
+			// for debug
+			log_debug("Get respose buf: %s", core_databuf_get_ptr(databuf));
+
+			nng_msg* msg;
+			size_t msg_size;
+			msg_size = msg_external_data_get_size();
+			rv = nng_msg_alloc(&msg, msg_size);
+			if (rv == 0) {
+				message_t* pay_msg;
+
+				pay_msg = (message_t*)nng_msg_body(msg);
+				msg_external_data_init(pay_msg, MSG_DATA_NEURON_DATABUF,
+					   				   databuf);
+				nng_sendmsg(adapter->sock, msg, 0);
+			}
+			core_databuf_put(databuf);
+			break;
+		}
+
+		default:
+			break;
+	}
 	return rv;
 }
 
-static int adapter_event_notify(neu_adapter_t* adapter, neu_event_notify_t* event)
+static int adapter_event_notify(neu_adapter_t* adapter,
+	   							neu_event_notify_t* event)
 {
 	int rv = 0;
 
@@ -190,12 +272,20 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info)
 		return NULL;
 	}
 
+	int rv;
+	adapter->state = ADAPTER_STATE_NULL;
+	adapter->stop  = false;
+	if ((rv = nng_mtx_alloc(&adapter->mtx)) != 0) {
+		log_error("Can't allocate mutex for adapter");
+		free(adapter);
+		return NULL;
+	}
+			
 	adapter->id = info->id;
 	adapter->type = info->type;
 	adapter->name = strdup(info->name);
 	adapter->plugin_lib_name = strdup(info->plugin_lib_name);
-	adapter->state = ADAPTER_STATE_IDLE;
-	adapter->stop  = false;
+	adapter->new_req_id = 0;
 
 	if (  adapter->name == NULL
 	   || adapter->plugin_lib_name == NULL) {
@@ -205,6 +295,7 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info)
 		if (adapter->plugin_lib_name != NULL) {
 			free(adapter->plugin_lib_name);
 		}
+		nng_mtx_free(adapter->mtx);
 		free(adapter);
 		log_error("Failed duplicate string for create adapter");
 		return NULL;
@@ -228,6 +319,11 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info)
 		neu_panic("Can't to open plugin(%s)", plugin_module->module_name);
 	}
 
+	rv = nng_pair1_open(&adapter->sock);
+	if (rv != 0) {
+		neu_panic("The adapter(%s) can't open pipe", adapter->name);
+	}
+
 	adapter->plugin = plugin;
 	return adapter;
 }
@@ -238,6 +334,7 @@ void neu_adapter_destroy(neu_adapter_t* adapter)
 		neu_panic("Destroied adapter is NULL");
 	}
 
+	nng_close(adapter->sock);
 	adapter->plugin_module->intf_funs->close(adapter->plugin);
 	unload_plugin_library(adapter->plugin_lib);
 	if (adapter->name != NULL) {
@@ -246,6 +343,7 @@ void neu_adapter_destroy(neu_adapter_t* adapter)
 	if (adapter->plugin_lib_name != NULL) {
 		free(adapter->plugin_lib_name);
 	}
+	nng_mtx_free(adapter->mtx);
 	free(adapter);
 	return;
 }
@@ -272,6 +370,10 @@ int neu_adapter_stop(neu_adapter_t* adapter, neu_manager_t* manager)
 {
 	int rv = 0;
 
+	log_info("Stop the adapter(%s)", adapter->name);
+	nng_mtx_lock(adapter->mtx);
+	adapter->stop = true;
+	nng_mtx_unlock(adapter->mtx);
 	nng_thread_destroy(adapter->thrd);
 
 	const neu_plugin_intf_funs_t* intf_funs;
@@ -281,3 +383,17 @@ int neu_adapter_stop(neu_adapter_t* adapter, neu_manager_t* manager)
 	return rv;
 }
 
+const char* neu_adapter_get_name(neu_adapter_t* adapter)
+{
+	return (const char*)adapter->name;
+}
+
+neu_manager_t* neu_adapter_get_manager(neu_adapter_t* adapter)
+{
+	return (neu_manager_t*)adapter->manager;
+}
+
+nng_socket neu_adapter_get_sock(neu_adapter_t* adapter)
+{
+	return adapter->sock;
+}

src/adapter/adapter_internal.h

@@ -20,6 +20,7 @@
 #ifndef ADAPTER_INTERNAL_H
 #define ADAPTER_INTERNAL_H
 
+#include "core/databuf.h"
 #include "core/neu_manager.h"
 
 typedef enum adapter_type {
@@ -42,5 +43,8 @@ neu_adapter_t* neu_adapter_create(neu_adapter_info_t* info);
 void neu_adapter_destroy(neu_adapter_t* adapter);
 int neu_adapter_start(neu_adapter_t* adapter, neu_manager_t* manager);
 int neu_adapter_stop(neu_adapter_t* adapter, neu_manager_t* manager);
+const char* neu_adapter_get_name(neu_adapter_t* adapter);
+neu_manager_t* neu_adapter_get_manager(neu_adapter_t* adapter);
+nng_socket neu_adapter_get_sock(neu_adapter_t* adapter);
 
 #endif

src/core/databuf.c

@@ -25,21 +25,23 @@
 
 struct core_databuf {
 	neu_atomic_int ref_count;
-	size_t	 len;
-	char     buf[0];
+	size_t	 	   len;
+	void*     	   buf;
 };
 
-core_databuf_t* core_databuf_alloc(size_t size)
+core_databuf_t* core_databuf_new_with_buf(void* buf, size_t len)
 {
 	core_databuf_t* databuf;
 	
-	databuf = (core_databuf_t*)malloc(sizeof(core_databuf_t) + size);
+	databuf = (core_databuf_t*)malloc(sizeof(core_databuf_t));
 	if (databuf == NULL) {
 		return NULL;
 	}
 
 	neu_atomic_init(&databuf->ref_count);
 	neu_atomic_set(&databuf->ref_count, 1);
+	databuf->len = len;
+	databuf->buf = buf;
 	return databuf;
 }
 
@@ -53,16 +55,21 @@ core_databuf_t* core_databuf_get(core_databuf_t* databuf)
 	return databuf;
 }
 
-void core_databuf_put(core_databuf_t* databuf)
+core_databuf_t* core_databuf_put(core_databuf_t* databuf)
 {
 	if (databuf == NULL) {
-		return;
+		return NULL;
 	}
 
 	if (neu_atomic_dec_nv(&databuf->ref_count) == 0) {
+		if (databuf->buf != NULL) {
+			free(databuf->buf);
+		}
 		free(databuf);
+		return NULL;
 	}
-	return;
+
+	return databuf;
 }
 
 size_t core_databuf_get_len(core_databuf_t* databuf)
@@ -74,7 +81,23 @@ size_t core_databuf_get_len(core_databuf_t* databuf)
 	return databuf->len;
 }
 
-void* core_databuf_peek_ptr(core_databuf_t* databuf)
+core_databuf_t* core_databuf_set_buf(core_databuf_t* databuf,
+									 void* buf, size_t len)
+{
+	if (databuf == NULL) {
+		return NULL;
+	}
+	
+	if (databuf->buf != NULL) {
+		free(databuf->buf);
+	}
+
+	databuf->len = len;
+	databuf->buf = buf;
+	return databuf;
+}
+
+void* core_databuf_get_ptr(core_databuf_t* databuf)
 {
 	if (databuf == NULL) {
 		return NULL;

src/core/databuf.h

@@ -22,17 +22,38 @@
 
 typedef struct core_databuf core_databuf_t;
 
+/**
+ * Create a shared databuf with provided buffer, the ownership of buffer will
+ * move into shared databuf.
+ * The caller don't free the buffer.
+ */
+core_databuf_t* core_databuf_new_with_buf(void* buf, size_t len);
+
 /**
  * Get a shared databuf, it will increase reference count of databuf
  */
 core_databuf_t* core_databuf_get(core_databuf_t* databuf);
 
 /**
- * Put a shared databuf, it will decrease reference count of databuf
+ * Put a shared databuf, it will decrease reference count of databuf.
+ * If the reference count of databuf is 0, then free databuf and buffer in it.
  */
-void core_databuf_put(core_databuf_t* databuf);
+core_databuf_t* core_databuf_put(core_databuf_t* databuf);
 
+/**
+ * Get length of the buffer in databuf
+ */
 size_t core_databuf_get_len(core_databuf_t* databuf);
-void* core_databuf_peek_ptr(core_databuf_t* databuf);
+
+/**
+ * Replace original buffer pointer in databuf with new buffer pointer, the
+ * original buffer should be free.
+ */
+core_databuf_t* core_databuf_set_buf(core_databuf_t* databuf,
+									 void* buf, size_t len);
+/**
+ * Get a buffer pointer in databuf
+ */
+void* core_databuf_get_ptr(core_databuf_t* databuf);
 
 #endif

src/core/message.c

@@ -37,7 +37,6 @@ struct message {
     union {
 		msg_inplace_buf_t    inp_buf;
 		// The external buffer pointer to a core_databuf_t.
-		// The one core_databuf_t is get from a buffer pool.
 		core_databuf_t*      ext_buf;
     };
 
@@ -144,7 +143,7 @@ void* msg_get_buf_ptr(message_t* msg)
 	}
 
 	if ((msg->type & MSG_DATABUF_KIND_MASK) == MSG_DATABUF_EXTERNAL) {
-		return core_databuf_peek_ptr(msg->ext_buf);
+		return core_databuf_get_ptr(msg->ext_buf);
 	} else {
 		return msg->inp_buf.buf_ptr;
 	}

src/core/message.h

@@ -28,13 +28,18 @@
 typedef enum msg_type {
 	MSG_TYPE_CMD_START = 0,
 	MSG_CMD_NOP = MSG_TYPE_CMD_START,
-	MSG_CMD_START_NEURON,
-	MSG_CMD_STOP_NEURON,
+	MSG_CMD_START_READ,
+	MSG_CMD_STOP_READ,
+	MSG_CMD_EXIT_LOOP,
 	MSG_TYPE_CMD_END,
 
 	MSG_TYPE_CONFIG_START = 0x1000,
-	MSG_CONFIG_SUB_NEURON_DRIVER = MSG_TYPE_CONFIG_START,
-	MSG_CONFIG_INFO_STRING,
+	MSG_CONFIG_INFO_STRING= MSG_TYPE_CONFIG_START,
+	MSG_CONFIG_REGISTER_ADAPTER,
+	MSG_CONFIG_UNREGISTER_ADAPTER,
+	MSG_CONFIG_SUBSCRIBE_ADAPTER,
+	// for adapter
+	MSG_CONFIG_READ_ADDR_RANGE,
 	MSG_TYPE_CONFIG_END,
 
 	MSG_TYPE_EVENT_START = 0x2000,
@@ -43,11 +48,11 @@ typedef enum msg_type {
 	MSG_TYPE_EVENT_END,
 
 	MSG_TYPE_DATA_START = 0x3000,
-	MSG_DATA_NEURON_DRIVER = MSG_TYPE_DATA_START,
+	MSG_DATA_NEURON_DATABUF = MSG_TYPE_DATA_START,
 	MSG_TYPE_DATA_END,
 
 	// User customer message types
-	MSG_TYPE_VENDOR_START = 0x10000,
+	MSG_TYPE_VENDOR_START = 0x100000,
 	MSG_TYPE_VENDOR_END,
 
 	MSG_TYPE_END = MSG_TYPE_VENDOR_END,

src/core/neu_manager.c

@@ -34,8 +34,8 @@
 
 typedef struct adapter_bind_info {
 	uint32_t adapter_id;
+	neu_adapter_t* adapter;
 	nng_pipe adapter_pipe;
-	int      adapter_sock;
 	int      bind_count;
 } adapter_bind_info_t;
 
@@ -52,8 +52,16 @@ typedef struct manager_bind_info {
 	adapter_bind_info_t* bind_adapters;
 } manager_bind_info_t;
 
+typedef enum manager_state {
+	MANAGER_STATE_NULL,
+	MANAGER_STATE_IDLE,
+	MANAGER_STATE_RUNNING,
+} manager_state_e;
+
 struct neu_manager {
 	const char*			listen_url;
+	nng_mtx*			mtx;
+	manager_state_e		state;
 	bool				stop;
 	nng_thread*			thrd;
 	uint32_t			new_adapter_id;
@@ -118,20 +126,97 @@ static int manager_add_config(neu_manager_t* manager)
 	return rv;
 }
 
-static int manager_add_adapter(neu_manager_t* manager, neu_adapter_t* adapter)
+static uint32_t manager_get_adapter_id(neu_manager_t* manager)
+{
+	uint32_t adapter_id;
+
+	adapter_id = manager->new_adapter_id++;
+	return adapter_id;
+}
+
+static void manager_bind_adapter(nng_pipe p,
+								 nng_pipe_ev ev,
+								 void* arg)
+{
+	int rv = 0;
+	neu_adapter_t* adapter;
+
+	adapter = (neu_adapter_t*)arg;
+	log_info("The manager will bind adapter(%s)",
+			 neu_adapter_get_name(adapter));
+
+	return;
+}
+
+static void manager_unbind_adapter(nng_pipe p,
+								   nng_pipe_ev ev,
+								   void* arg)
+{
+	int rv = 0;
+	neu_adapter_t* adapter;
+
+	adapter = (neu_adapter_t*)arg;
+	log_info("The manager will unbind adapter(%s)",
+			 neu_adapter_get_name(adapter));
+
+	return;
+}
+
+static int manager_add_adapter(neu_manager_t* manager,
+	   						   neu_adapter_t* adapter)
 {
 	int rv = 0;
 
 	return rv;
 }
 
-static int manager_remove_adapter(neu_manager_t* manager, neu_adapter_t* adapter)
+static int manager_remove_adapter(neu_manager_t* manager,
+	   							  neu_adapter_t* adapter)
 {
 	int rv = 0;
 
 	return rv;
 }
 
+static int manager_start_adapter(neu_manager_t* manager,
+								neu_adapter_t* adapter)
+{
+	int rv = 0;
+
+	rv = nng_pipe_notify(neu_adapter_get_sock(adapter),
+						 NNG_PIPE_EV_ADD_POST,
+						 manager_bind_adapter,
+						 adapter);
+	neu_adapter_start(adapter, manager);
+	return rv;
+}
+
+static int manager_stop_adapter(neu_manager_t* manager,
+								neu_adapter_t* adapter)
+{
+	int rv = 0;
+	size_t   msg_size;
+	nng_msg* msg;
+
+	msg_size = msg_inplace_data_get_size(sizeof(uint32_t));
+	rv = nng_msg_alloc(&msg, msg_size);
+	if (rv != 0) {
+		neu_panic("Can't allocate msg for stop adapter(%s)",
+			      neu_adapter_get_name(adapter));
+	}
+
+	void*  buf_ptr;
+	message_t* pay_msg;
+	pay_msg = (message_t*)nng_msg_body(msg);
+	msg_inplace_data_init(pay_msg, MSG_CMD_EXIT_LOOP, sizeof(uint32_t));
+	buf_ptr = msg_get_buf_ptr(pay_msg);
+	*(uint32_t*)buf_ptr = 0;	// exit_code is 0
+	manager_bind_info_t* manager_bind = &manager->bind_info;
+	nng_sendmsg(manager_bind->mng_sock, msg, 0);
+	neu_adapter_stop(adapter, manager);
+	return rv;
+}
+
 static void manager_loop(void* arg)
 {
 	int rv;
@@ -149,33 +234,41 @@ static void manager_loop(void* arg)
 		neu_panic("Neuron manager can't listen on %s", manager->listen_url);
 	}
 
+	nng_mtx_lock(manager->mtx);
+	manager->state = MANAGER_STATE_RUNNING;
+	nng_mtx_unlock(manager->mtx);
+
 #ifdef NEU_HAS_SAMPLE_ADAPTER
-	sample_adapter_info.id = manager->new_adapter_id;
-	manager->new_adapter_id++;
+	sample_adapter_info.id = manager_get_adapter_id(manager);
 	sample_adapter = neu_adapter_create(&sample_adapter_info);
-	neu_adapter_start(sample_adapter, manager);
+	manager_start_adapter(manager, sample_adapter);
 #endif
 
 	while (1) {
 		nng_msg* msg;
 
+		nng_mtx_lock(manager->mtx);
+		if (manager->stop) {
+			manager->state = MANAGER_STATE_NULL;
+			nng_mtx_unlock(manager->mtx);
+			log_info("Exit loop of the manager");
+			break;
+		}
+		nng_mtx_unlock(manager->mtx);
+
 		rv = nng_recvmsg(manager_bind->mng_sock, &msg, 0);
 		if (rv != 0) {
-			if (manager->stop) {
-				break;
-			} else {
-				log_warn("Manage pipe no message received");
-				continue;
-			}
+			log_warn("Manage pipe no message received");
+			continue;
 		}
 
-		message_t* neu_msg;
-		neu_msg = nng_msg_body(msg);
-		switch (msg_get_type(neu_msg)) {
+		message_t* pay_msg;
+		pay_msg = nng_msg_body(msg);
+		switch (msg_get_type(pay_msg)) {
 			case MSG_EVENT_STATUS_STRING:
 			{
 				char* buf_ptr;
-				buf_ptr = msg_get_buf_ptr(neu_msg);
+				buf_ptr = msg_get_buf_ptr(pay_msg);
 				log_info("Recieve string: %s", buf_ptr);
 
 				const char* adapter_str = "manager recv reply";
@@ -184,21 +277,65 @@ static void manager_loop(void* arg)
 				msg_size = msg_inplace_data_get_size(strlen(adapter_str) + 1);
 				rv = nng_msg_alloc(&out_msg, msg_size);
 				if (rv == 0) {
-					void* msg_ptr;
+					message_t* msg_ptr;
 					char* buf_ptr;
-					msg_ptr = nng_msg_body(out_msg);
+					msg_ptr = (message_t*)nng_msg_body(out_msg);
 					msg_inplace_data_init(msg_ptr, MSG_CONFIG_INFO_STRING,
-							msg_size);
+							strlen(adapter_str) + 1);
 					buf_ptr = msg_get_buf_ptr(msg_ptr);
 					memcpy(buf_ptr, adapter_str, strlen(adapter_str));
 					buf_ptr[strlen(adapter_str)] = 0;
 					nng_sendmsg(manager_bind->mng_sock, out_msg, 0);
 				}
+
+				// for debug
+				usleep(1000);
+				nng_msg* read_msg;
+				msg_size = msg_inplace_data_get_size(sizeof(uint32_t));
+				rv = nng_msg_alloc(&read_msg, msg_size);
+				if (rv == 0) {
+					message_t* msg_ptr;
+					void* buf_ptr;
+
+					msg_ptr = (message_t*)nng_msg_body(read_msg);
+					msg_inplace_data_init(msg_ptr, MSG_CMD_START_READ,
+										  sizeof(uint32_t));
+					buf_ptr = msg_get_buf_ptr(msg_ptr);
+					*(uint32_t*)buf_ptr = 3;
+					nng_sendmsg(manager_bind->mng_sock, read_msg, 0);
+				}
+				break;
+			}
+
+			case MSG_CMD_EXIT_LOOP:
+			{
+				uint32_t exit_code;
+
+				exit_code = *(uint32_t*)msg_get_buf_ptr(pay_msg);
+				log_info("manager exit loop by exit_code=%d", exit_code);
+				nng_mtx_lock(manager->mtx);
+				manager->state = MANAGER_STATE_NULL;
+				manager->stop  = true;
+				nng_mtx_unlock(manager->mtx);
+				break;
+			}
+
+			case MSG_DATA_NEURON_DATABUF:
+			{
+				core_databuf_t* databuf;
+
+				databuf = (core_databuf_t*)msg_get_buf_ptr(pay_msg);
+
+				log_info("dispatch databuf to subscribes");
+				// TODO: dispatch the databuf to subsribes
+
+				core_databuf_put(databuf);
 				break;
 			}
 
 			default:
-				log_warn("Receive a not handled message(type: %d)", msg_get_type(neu_msg));
+				log_warn("Receive a not supported message(type: %d)",
+					     msg_get_type(pay_msg));
 				break;
 		}
 
@@ -206,9 +343,10 @@ static void manager_loop(void* arg)
 	}
 
 #ifdef NEU_HAS_SAMPLE_ADAPTER
-	neu_adapter_stop(sample_adapter, manager);
+	manager_stop_adapter(manager, sample_adapter);
 	neu_adapter_destroy(sample_adapter);
 #endif
+	nng_close(manager_bind->mng_sock);
 	return;
 }
 
@@ -222,9 +360,17 @@ neu_manager_t* neu_manager_create()
 		return NULL;
 	}
 
+	manager->state		= MANAGER_STATE_NULL;
 	manager->stop		= false;
 	manager->listen_url = manager_url;
 	manager->new_adapter_id = 1;
+	int rv;
+	if ((rv = nng_mtx_alloc(&manager->mtx)) != 0) {
+		log_error("Can't allocate mutex for manager");
+		free(manager);
+		return NULL;
+	}
+
 	init_bind_info(&manager->bind_info);
 	nng_thread_create(&manager->thrd, manager_loop, manager);
 	return manager;
@@ -233,35 +379,12 @@ neu_manager_t* neu_manager_create()
 void neu_manager_destroy(neu_manager_t* manager)
 {
 	nng_thread_destroy(manager->thrd);
+	nng_mtx_free(manager->mtx);
 	uninit_bind_info(&manager->bind_info);
 	free(manager);
 	return;
 }
 
-int neu_manager_bind_adapter(neu_manager_t* manager, neu_adapter_t* adapter)
-{
-	int rv = 0;
-
-	if (adapter == NULL) {
-		log_warn("The adapter is NULL that want to bind");
-		return (-1);
-	}
-
-	return rv;
-}
-
-int neu_manager_unbind_adapter(neu_manager_t* manager, neu_adapter_t* adapter)
-{
-	int rv = 0;
-
-	if (adapter == NULL) {
-		log_warn("The adapter is NULL that want to unbind");
-		return (-1);
-	}
-
-	return rv;
-}
-
 const char* neu_manager_get_url(neu_manager_t* manager)
 {
 	if (manager == NULL) {

src/core/neu_manager.h

@@ -28,8 +28,6 @@ typedef struct neu_manager neu_manager_t;
 neu_manager_t* neu_manager_create();
 void neu_manager_destroy(neu_manager_t* manager);
 
-int neu_manager_bind_adapter(neu_manager_t* manager, neu_adapter_t* adapter);
-int neu_manager_unbind_adapter(neu_manager_t* manager, neu_adapter_t* adapter);
 const char* neu_manager_get_url(neu_manager_t* manager);
 
 #endif