update

wangzhengquan

2020-12-24 dbf7fd6f620b86c382431ce10fd5fe3c5ddca955

 src/socket/bus_server_socket.c

@@ -1,6 +1,7 @@

#include "bus_server_socket.h"
#include "shm_mod_socket.h"
#include "bus_error.h"

static Logger *logger = LoggerFactory::getLogger();

@@ -60,15 +61,19 @@


BusServerSocket::BusServerSocket() {
   logger->debug("BusServerSocket Init");
   shm_socket = shm_open_socket(SHM_SOCKET_DGRAM);
   topic_sub_map = NULL;

}

BusServerSocket::~BusServerSocket() {
   SHMKeySet *subscripter_set;
   SHMTopicSubMap::iterator map_iter;

   logger->debug("BusServerSocket destory 1");
   stop();
   logger->debug("BusServerSocket destory 2");
    
   if(topic_sub_map != NULL) {
      for (map_iter = topic_sub_map->begin(); map_iter != topic_sub_map->end(); map_iter++) {
@@ -83,6 +88,7 @@
      mem_pool_free_by_key(BUS_MAP_KEY);
   }
   shm_close_socket(shm_socket);
   logger->debug("BusServerSocket destory 3");
}


@@ -109,14 +115,13 @@
 
   run_pubsub_proxy();
   // 进程停止的时候，预留3秒资源回收的时间。否则，会发生调用close的时候，共享内存的资源还没来得及回收进程就退出了
   sleep(3);
   return 0;
}


int  BusServerSocket::stop(){
   int ret;
	 
   logger->debug("====>stopping");
   if( shm_socket->key <= 0) {
      return -1;
   }
@@ -127,15 +132,11 @@
   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);
      ret = shm_sendandrecv_unsafe(shm_socket, buf, size, shm_socket->key, NULL, NULL);
      free(buf);
      free(recv_buf);
      return ret;
   } else {
      return -1;
@@ -212,7 +213,7 @@
      for(set_iter = subscripter_set->begin(); set_iter != subscripter_set->end(); set_iter++) {
         send_key = *set_iter;
 // printf("_proxy_pub send before %d \n", send_key);
         if (shm_sendto(shm_socket, buf, size, send_key, &timeout) == SHM_SOCKET_ECONNFAILED ) {
         if (shm_sendto(shm_socket, buf, size, send_key, &timeout) == EBUS_CLOSED ) {
            //对方已关闭的连接放到待删除队列里。如果直接删除会让iter指针出现错乱
            subscripter_to_del.push_back(send_key);
         } else {
@@ -260,7 +261,8 @@
        topic =  strtok(NULL, topic_delim);
        }

      } else if(strcmp(action, "desub") == 0) {
      } 
      else if(strcmp(action, "desub") == 0) {
// printf("desub topic=%s,%s,%d\n", topics, trim(topics, 0), strcmp(trim(topics, 0), ""));
         if(strcmp(trim(topics, 0), "") == 0) {
            // 取消所有订阅
@@ -274,27 +276,26 @@
           }
         }
         
      } else if(strcmp(action, "pub") == 0) {
      } 
      else if(strcmp(action, "pub") == 0) {
          content = topics + head.topic_size;
         _proxy_pub(topics, content, head.content_size, key);
      }  else if(strcmp(action, "stop") == 0) {
         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, "stop_finished", strlen( "stop_finished") +1, key);

      }  
      else if(strcmp(action, "stop") == 0) {
			
         free(buf);
         break;
      } else {
         logger->error( "BusServerSocket::run_pubsub_proxy : unrecognized action %s", action);
      }
      
      // free(action);
      // free(topics);
      // } else {
      //    logger->error( "BusServerSocket::run_pubsub_proxy : incorrect format msg");
      // }
      free(buf);
   }

   logger->info( "Stopping Bus...");
   shm_sendto(shm_socket, "stop_finished", strlen( "stop_finished") +1, key);

   return NULL;
}