解决sendandrecv发送到一个不存在key的情况

wangzhengquan

2020-12-22 26ed48c4e616014ee760fd13d13dbdc8539c34e3

 src/socket/net_mod_socket.c

@@ -15,7 +15,8 @@

NetModSocket::NetModSocket() 
{
  if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR)    err_msg(errno, "signal");
  if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR)
      logger->error(errno, "NetModSocket::NetModSocket signal");
}


@@ -56,21 +57,23 @@


/* Free thread-specific data buffer */
void NetModSocket::_destructor_(void *_pool)
void NetModSocket::_destroyConnPool_(void *_pool)
{

  NetConnPool *mpool = (NetConnPool *)_pool;
  delete mpool;
  logger->debug("destory connPool");
}

 /* One-time key creation function */
void NetModSocket::_createKey_(void)
void NetModSocket::_createConnPoolKey_(void)
{
  int ret;

  /* Allocate a unique thread-specific data key and save the address
     of the destructor for thread-specific data buffers */

  ret = pthread_key_create(&poolKey, _destructor_);
  ret = pthread_key_create(&poolKey, _destroyConnPool_);
  if (ret != 0) {
    logger->error(ret, "pthread_key_create");
    exit(1);
@@ -82,9 +85,7 @@

  int i, n, recv_size, connfd;
  net_node_t *node;
  void *recv_buf;

 
  void *recv_buf = NULL;
  
  net_mod_request_head_t request_head = {};
 
@@ -97,8 +98,7 @@
  NetConnPool *mpool;

  /* Make first caller allocate key for thread-specific data */

  ret = pthread_once(&once, _createKey_);
  ret = pthread_once(&once, _createConnPoolKey_);
  if (ret != 0) {
    LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_once");
    exit(1);
@@ -107,8 +107,8 @@
  mpool = (NetConnPool *)pthread_getspecific(poolKey);
  if (mpool == NULL)
  {
    /* If first call from this thread, allocate
                                   buffer for thread, and save its location */
    /* If first call from this thread, allocate buffer for thread, and save its location */
    logger->debug("Create connPool");
    mpool = new NetConnPool();
    if (mpool == NULL) {
      LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ malloc");
@@ -129,15 +129,17 @@
  for (i = 0; i< arrlen; i++) {

    node = &node_arr[i];
    if(node->host == NULL) {
    if(node->host == NULL || strcmp(node->host, "") == 0 ) {
      // 本地发送
      shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size);
      strcpy( ret_arr[n_recv_suc].host,"localshm");
      ret_arr[n_recv_suc].port = 0;
      ret_arr[n_recv_suc].key = node->key;
      ret_arr[n_recv_suc].content = recv_buf;
      ret_arr[n_recv_suc].content_length = recv_size;
      n_recv_suc++;
      if(shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size) == 0) {
        strcpy( ret_arr[n_recv_suc].host,"");
        ret_arr[n_recv_suc].port = 0;
        ret_arr[n_recv_suc].key = node->key;
        ret_arr[n_recv_suc].content = recv_buf;
        ret_arr[n_recv_suc].content_length = recv_size;
        n_recv_suc++;
      }
     
      continue;
    }

@@ -235,6 +237,17 @@
     
}


void NetModSocket::free_recv_msg_arr(net_mod_recv_msg_t * arr, size_t size) {

  for(int i =0; i< size; i++) {
    if(arr[i].content != NULL)
      free(arr[i].content);
  }
  free(arr);
}


int NetModSocket::pub(net_node_t *node_arr, int arrlen, char *topic, int topic_size, void *content, int content_size) {
  return _pub_(node_arr, arrlen, topic, topic_size, content,   content_size, -1);
}
@@ -264,7 +277,7 @@
  NetConnPool *mpool;

  /* Make first caller allocate key for thread-specific data */
  ret = pthread_once(&once, _createKey_);
  ret = pthread_once(&once, _createConnPoolKey_);
  if (ret != 0) {
    LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_once");
    exit(1);
@@ -287,7 +300,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];
@@ -303,6 +322,7 @@
        continue;
      }
      request_head.mod = BUS;
      memcpy(request_head.host, node->host, sizeof(request_head.host));
      request_head.key = node->key;
      request_head.content_length = content_size;
      request_head.topic_length = strlen(topic) + 1;
@@ -559,15 +579,6 @@


/**
 * 启动bus
 * 
 * @return 0 成功， 其他值 失败的错误码
*/
int  NetModSocket::start_bus() {
  return shmModSocket.start_bus();
}

/**
 * 订阅指定主题
 * @topic 主题
 * @size 主题长度
@@ -633,14 +644,6 @@
  return shmModSocket.get_key();
}


void NetModSocket::free_recv_msg_arr(net_mod_recv_msg_t * arr, size_t size) {

  for(int i =0; i< size; i++) {
    free(arr[i].content);
  }
  free(arr);
}