update

wangzhengquan

2020-08-06 9b9220321e647b381a999f67cad12345334b2cbe

 src/socket/dmod_socket.c

@@ -19,15 +19,50 @@
   }
}

bool DModSocket::include_in_keys(int key, int keys[], size_t length) {
   if(length == 0) {
      return false;
   }
   for(int i = 0; i < length; i++) {
      if(keys[i] == key) 
         return true;
   }
   return false;
}

size_t DModSocket::remove_subscripters(int keys[], size_t length) {
   size_t count;
   foreach_subscripters([keys, length, &count](SHMKeySet *subscripter_set, SHMKeySet::iterator set_iter){
      if (include_in_keys(*set_iter, keys, length)) {
         subscripter_set->erase(set_iter);
         count++;
      }
   });
   return count;
}


size_t DModSocket::remove_keys(int keys[], size_t length) {
   remove_subscripters(keys, length);
   return shm_socket_remove_keys(keys, length);
}

DModSocket::DModSocket() {
      shm_socket = shm_open_socket(SHM_SOCKET_DGRAM);
      bus_set = new std::set<int>;
}

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

   struct timespec timeout = {1, 0};
   if(bus_set != NULL) {
      for(auto bus_iter = bus_set->begin(); bus_iter != bus_set->end(); bus_iter++) {
         desub_timeout(NULL, 0, *bus_iter, &timeout);
      }
      delete bus_set;
   }
	
   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;
@@ -135,15 +170,7 @@
   // pthread_create(&tid, NULL, run_accept_sub_request, _socket);
   return 0;
}
/**
 * @port 总线端口
 */
int  DModSocket::_sub_( void *topic, int size, int port,  
   struct timespec *timeout, int flags) {
   char buf[8192];
   snprintf(buf,  8192, "%ssub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
   return shm_sendto(shm_socket, buf, strlen(buf) + 1, port, timeout, flags);
}

/**
 * 订阅指定主题
 * @topic 主题
@@ -162,19 +189,25 @@
}



/**
 * 取消订阅指定主题
 * @topic 主题
 * @size 主题长度
 * @port 总线端口
 */
int  DModSocket::_pub_( void *topic, int topic_size, void *content, int content_size, int port,  
   struct timespec *timeout, int flags) {
   int head_len;
   char buf[8192+content_size];
   snprintf(buf, 8192, "%spub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
   head_len = strlen(buf);
   memcpy(buf+head_len, content, content_size);
   return shm_sendto(shm_socket, buf, head_len+content_size, port, timeout, flags);

int  DModSocket::desub( void *topic, int size, int port){
   return _desub_( topic, size, port, NULL, 0);
}
// 超时返回。 @sec 秒 ， @nsec 纳秒
int  DModSocket::desub_timeout(void *topic, int size, int port, struct timespec *timeout){
   return _desub_(topic, size, port, timeout, 0);
}
int  DModSocket::desub_nowait(void *topic, int size, int port) {
   return _desub_(topic, size, port, NULL,  (int)SHM_MSG_NOWAIT);
}



/**
 * 发布主题
@@ -203,9 +236,46 @@



// ========================================================
// =============================================================================
/**
 * @port 总线端口
 */
int  DModSocket::_sub_( void *topic, int size, int port,  
   struct timespec *timeout, int flags) {
   char buf[8192];
   int rv;
   snprintf(buf,  8192, "%ssub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
   rv = shm_sendto(shm_socket, buf, strlen(buf) + 1, port, timeout, flags);
   if(rv == 0) {
      bus_set->insert(port);
   }
   return rv;
}


/**
 * @port 总线端口
 */
int  DModSocket::_desub_( void *topic, int size, int port,  
   struct timespec *timeout, int flags) {
   char buf[8192];
   snprintf(buf,  8192, "%sdesub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
   return shm_sendto(shm_socket, buf, strlen(buf) + 1, port, timeout, flags);
}

/**
 * @port 总线端口
 */
int  DModSocket::_pub_( void *topic, int topic_size, void *content, int content_size, int port,  
   struct timespec *timeout, int flags) {
   int head_len;
   char buf[8192+content_size];
   snprintf(buf, 8192, "%spub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
   head_len = strlen(buf);
   memcpy(buf+head_len, content, content_size);
   return shm_sendto(shm_socket, buf, head_len+content_size, port, timeout, flags);

}
/*
 * 处理订阅
*/
@@ -223,6 +293,35 @@
      topic_sub_map->insert({topic, subscripter_set});
   }
   subscripter_set->insert(port);
}

/*
 * 处理取消订阅
*/
void DModSocket::_proxy_desub( char *topic, int port) {
   SHMKeySet *subscripter_set;

   SHMTopicSubMap::iterator map_iter;
   // SHMKeySet::iterator set_iter;

   if( (map_iter = topic_sub_map->find(topic) ) != topic_sub_map->end()) {
      subscripter_set = map_iter->second;
      subscripter_set->erase(port);
   }
}

/*
 * 处理取消所有订阅
*/
void DModSocket::_proxy_desub_all(int port) {
   SHMKeySet *subscripter_set;

   SHMTopicSubMap::iterator map_iter;
   // SHMKeySet::iterator set_iter;
   for (auto map_iter = topic_sub_map->begin(); map_iter != topic_sub_map->end(); map_iter++) {
         subscripter_set = map_iter->second;
         subscripter_set->erase(port);
   }
}

/*
@@ -281,15 +380,31 @@
      if(parse_pubsub_topic(buf, size, &action, &topics, &head_len)) {
         if(strcmp(action, "sub") == 0) {
            // 订阅支持多主题订阅
            topic = trim(strtok(topics, topic_delim), NULL);
            topic = strtok(topics, topic_delim);
           while(topic) {
          _proxy_sub( topic, port);
           topic = trim(strtok(NULL, topic_delim), NULL);
          _proxy_sub(trim(topic, 0), port);
           topic =  strtok(NULL, topic_delim);
           }

         } else if(strcmp(action, "desub") == 0) {
            if(strcmp(trim(topics, 0), "") == 0) {
               // 取消所有订阅
      printf("取消所有订阅");
               _proxy_desub_all(port);
            } else {
				 
               topic = strtok(topics, topic_delim);
              while(topic) {
             _proxy_desub(trim(topic, 0), port);
              topic =  strtok(NULL, topic_delim);
              }
            }
				
				

         } else if(strcmp(action, "pub") == 0) {
            _proxy_pub(topics, head_len, buf, size, port);
         }
         }  
         
         free(action);
         free(topics);
@@ -356,12 +471,14 @@
  return 0;
 }

 char *topic = (char *)calloc(1, topic_len+1);
 char *topic = (char *)malloc(topic_len+1);
 strncpy(topic, topic_start_ptr, topic_len);
 *topic = '\0';
 *_topic = topic;

 char *action = (char *)calloc(1, action_len+1);
 char *action = (char *)malloc(action_len+1);
 strncpy(action, action_start_ptr, action_len);
 *action = '\0';
 *_action = action;
 *head_len = ptr-str;