add desub

wangzhengquan

2020-08-06 203df24a403a8c0cd8e93d0f33eaf10de2788969

 src/socket/dmod_socket.c

@@ -135,15 +135,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 +154,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 +201,41 @@



// ========================================================
// =============================================================================
/**
 * @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);
}


/**
 * @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 +253,22 @@
      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);
   }
 
}

/*
@@ -281,10 +327,18 @@
      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) {
            // 订阅支持多主题订阅
            topic = strtok(topics, topic_delim);
           while(topic) {
          _proxy_desub(trim(topic, 0), port);
           topic =  strtok(NULL, topic_delim);
           }

         } else if(strcmp(action, "pub") == 0) {