update

wangzhengquan

2020-12-23 1b94589dcb8d497d2d8a208efd61a54631f6b84e

 src/socket/bus_server_socket.c

@@ -60,15 +60,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 +87,7 @@
      mem_pool_free_by_key(BUS_MAP_KEY);
   }
   shm_close_socket(shm_socket);
   logger->debug("BusServerSocket destory 3");
}


@@ -109,14 +114,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 +131,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;
@@ -260,7 +260,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 +275,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;
}