From b029ff78a59bfe8af4e66f844644a776f7678eef Mon Sep 17 00:00:00 2001
From: wangzhengquan <wangzhengquan85@126.com>
Date: 星期一, 21 十二月 2020 17:49:51 +0800
Subject: [PATCH] bus 加停止功能

---
 src/socket/net_mod_socket.c           |    8 +++
 src/socket/shm_mod_socket.h           |    2 +
 test_net_socket/test_net_mod_socket.c |    2 
 src/socket/bus_server_socket.c        |   56 ++++++++++++++++-----------
 src/socket/net_mod_server_socket.c    |    1 
 5 files changed, 43 insertions(+), 26 deletions(-)

diff --git a/src/socket/bus_server_socket.c b/src/socket/bus_server_socket.c
index e1ddb89..8ad7379 100644
--- a/src/socket/bus_server_socket.c
+++ b/src/socket/bus_server_socket.c
@@ -65,33 +65,24 @@
 }
 
 BusServerSocket::~BusServerSocket() {
-// printf("BusServerSocket  destory 1\n");
 	SHMKeySet *subscripter_set;
 	SHMTopicSubMap::iterator map_iter;
 
 	stop();
-	sleep(2);
-// printf("BusServerSocket  destory 2\n");	
+	 
 	if(topic_sub_map != NULL) {
 		for (map_iter = topic_sub_map->begin(); map_iter != topic_sub_map->end(); map_iter++) {
 			subscripter_set = map_iter->second;
-// printf("BusServerSocket  destory 2-1\n");	
 			if(subscripter_set != NULL) {
-// printf("BusServerSocket  destory 2-2\n");	
 				subscripter_set->clear();
-// printf("BusServerSocket  destory 2-3\n");	
 				mm_free((void *)subscripter_set);
-// printf("BusServerSocket  destory 2-4\n");	
 			}
 
 		}
 		topic_sub_map->clear();
 		mem_pool_free_by_key(BUS_MAP_KEY);
 	}
-// printf("BusServerSocket  destory 3\n");	
-	// printf("=============close socket\n");
 	shm_close_socket(shm_socket);
-// printf("BusServerSocket  destory 4\n");	
 }
 
 
@@ -107,6 +98,7 @@
 int BusServerSocket::force_bind(int key) {
 	return shm_socket_force_bind(shm_socket, key);
 }
+
 /**
  * 鍚姩bus
  * 
@@ -116,20 +108,39 @@
 	topic_sub_map =	mem_pool_attach<SHMTopicSubMap>(BUS_MAP_KEY);
  
 	run_pubsub_proxy();
-	// pthread_t tid;
-	// pthread_create(&tid, NULL, run_accept_sub_request, _socket);
+	// 杩涚▼鍋滄鐨勬椂鍊欙紝棰勭暀3绉掕祫婧愬洖鏀剁殑鏃堕棿銆傚惁鍒欙紝浼氬彂鐢熻皟鐢╟lose鐨勬椂鍊欙紝鍏变韩鍐呭瓨鐨勮祫婧愯繕娌℃潵寰楀強鍥炴敹杩涚▼灏遍��鍑轰簡
+	sleep(3);
 	return 0;
 }
 
 
 int  BusServerSocket::stop(){
-	char buf[128];
+	int ret;
 	 
 	if( shm_socket->key <= 0) {
 		return -1;
 	}
-	snprintf(buf, 128, "%sstop%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, "", TOPIC_RIDENTIFIER);
-	return shm_sendto(shm_socket, buf, strlen(buf), shm_socket->key, NULL, 0);
+	// snprintf(buf, 128, "%sstop%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, "", TOPIC_RIDENTIFIER);
+	// return shm_sendto(shm_socket, buf, strlen(buf), shm_socket->key, NULL, 0);
+	bus_head_t head = {};
+	memcpy(head.action, "stop", sizeof(head.action));
+	head.topic_size = 0;
+	head.content_size = 0;
+
+	void *recv_buf;
+	int recv_size;
+
+	void *buf;
+	int size = ShmModSocket::get_bus_sendbuf(head, NULL, 0, NULL,  0, &buf);
+	if(size > 0) {
+		ret = shm_sendandrecv(shm_socket, buf, size, shm_socket->key, &recv_buf, &recv_size);
+		free(buf);
+		free(recv_buf);
+		return ret;
+	} else {
+		return -1;
+	}
+
 }
 
 /*
@@ -239,8 +250,7 @@
 		head = ShmModSocket::decode_bus_head(buf);
 		topics = buf + BUS_HEAD_SIZE;
 		action = head.action;
-		// if(parse_pubsub_topic(buf, size, &action, &topics, &head_len)) {
-  printf("run_pubsub_proxy : %s, %s \n", action, topics);
+  // printf("run_pubsub_proxy : %s, %s \n", action, topics);
 		if(strcmp(action, "sub") == 0) {
 			// 璁㈤槄鏀寔澶氫富棰樿闃�
 			topic = strtok(topics, topic_delim);
@@ -268,13 +278,12 @@
 			 content = topics + head.topic_size;
 			_proxy_pub(topics, content, head.content_size, key);
 		}  else if(strcmp(action, "stop") == 0) {
-			 logger->info( "Stopping Bus...");
+			logger->info( "Stopping Bus...");
 			 // snprintf(resp_buf, 128, "%sstop_finished%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, "", TOPIC_RIDENTIFIER);
-			 // shm_sendto(shm_socket, resp_buf, strlen(resp_buf), key);
-			 // free(action);
-			 // free(topics);
-			 free(buf);
-			 break;
+			shm_sendto(shm_socket, "stop_finished", strlen( "stop_finished") +1, key);
+
+			free(buf);
+			break;
 		} else {
 			logger->error( "BusServerSocket::run_pubsub_proxy : unrecognized action");
 		}
@@ -292,6 +301,7 @@
 
 
 /**
+ * deprecate
  * @str "<**sub**>{缁忔祹}"
  */
 
diff --git a/src/socket/net_mod_server_socket.c b/src/socket/net_mod_server_socket.c
index 56f5907..0de0523 100644
--- a/src/socket/net_mod_server_socket.c
+++ b/src/socket/net_mod_server_socket.c
@@ -244,7 +244,6 @@
     else if(request_head.timeout == -1) {
       ret = shmModSocket.pub((char*)topic_buf, request_head.topic_length, buf, request_head.content_length, BUS_KEY);
     }
-printf("bus server pub ret=%d\n", ret);   
     response_head.code = ret;
     response_head.content_length = 0;
     if( rio_writen(connfd, NetModSocket::encode_response_head(response_head), NET_MODE_RESPONSE_HEAD_LENGTH) != NET_MODE_RESPONSE_HEAD_LENGTH )
diff --git a/src/socket/net_mod_socket.c b/src/socket/net_mod_socket.c
index 675c80b..1e1fc27 100644
--- a/src/socket/net_mod_socket.c
+++ b/src/socket/net_mod_socket.c
@@ -287,7 +287,13 @@
     }
   }
 
-  
+  // 鏈湴鍙戦��
+  if(node_arr == NULL || arrlen == 0) {
+    if(shmModSocket.pub(topic, topic_size, content, content_size, node->key) == 0 ) {
+      n_pub_suc++;
+    }
+  }
+
   for (i = 0; i < arrlen; i++) {
 
     node = &node_arr[i];
diff --git a/src/socket/shm_mod_socket.h b/src/socket/shm_mod_socket.h
index 9ec3044..4c82497 100644
--- a/src/socket/shm_mod_socket.h
+++ b/src/socket/shm_mod_socket.h
@@ -12,6 +12,7 @@
 #include "socket_def.h"
 
 #define BUS_HEAD_SIZE (64 + 2 * sizeof(uint32_t))
+class BusServerSocket;
 
 struct bus_head_t
 {
@@ -22,6 +23,7 @@
 
 
 class ShmModSocket {
+friend class BusServerSocket;
 private:
 	shm_socket_t *shm_socket;
   socket_mod_t mod;
diff --git a/test_net_socket/test_net_mod_socket.c b/test_net_socket/test_net_mod_socket.c
index 1b6cf33..2277182 100644
--- a/test_net_socket/test_net_mod_socket.c
+++ b/test_net_socket/test_net_mod_socket.c
@@ -83,7 +83,7 @@
   void * server_socket = bus_server_socket_wrapper_open();
   pthread_t tid;
   // 鍒涘缓涓�涓嚎绋�,鍙互鍏抽棴bus
-  pthread_create(&tid, NULL, bus_handler, server_socket);
+  // pthread_create(&tid, NULL, bus_handler, server_socket);
   if(bus_server_socket_wrapper_start_bus(server_socket) != 0) {
     printf("start bus failed\n");
     exit(1);

--
Gitblit v1.8.0