update

wangzhengquan

2020-10-13 2a4e4619f34a742e36693e589e0431347a72979b

 src/socket/net_mod_socket.c

@@ -12,7 +12,13 @@


NetModSocket::~NetModSocket() {
  
  rio_t * rio;
  for (auto map_iter = connectionMap.begin(); map_iter != connectionMap.end(); map_iter++) {
    rio = map_iter->second;
    if(rio != NULL) {
      free(rio);
    }
  }
}

int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
@@ -25,7 +31,7 @@
  void *recv_buf;
  int recv_size;
  char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH];
  net_mod_request_head_t request_head;
  net_mod_request_head_t request_head = {};
  net_mod_response_head_t response_head;
  std::map<std::string, rio_t*>::iterator mapIter;
  rio_t *rio;
@@ -54,28 +60,34 @@
    request_head.mod = REQ_REP;
    request_head.key = node->key;
    request_head.content_length = send_size;
    request_head.topic_length = 0;
    if(rio_writen(rio->rio_fd, NetModSocket::encode_request_head(request_head), NET_MODE_REQUEST_HEAD_LENGTH) != NET_MODE_REQUEST_HEAD_LENGTH) {
      err_exit(errno, "NetModSocket::send head rio_writen");
      LoggerFactory::getLogger()->error(errno, "NetModSocket::send head rio_writen");
      exit(1);

    }

    if(rio_writen(rio->rio_fd, send_buf, send_size) != send_size ) {
       err_exit(errno, "NetModSocket::send conent rio_writen");
      LoggerFactory::getLogger()->error(errno, "NetModSocket::send conent rio_writen");
      exit(1);
    }


    if ( rio_readnb(rio, response_head_bs, NET_MODE_RESPONSE_HEAD_LENGTH) !=  NET_MODE_RESPONSE_HEAD_LENGTH) {
      err_exit(errno, "NetModSocket::send  rio_readnb");
      LoggerFactory::getLogger()->error(errno, "NetModSocket::send  rio_readnb");
      exit(1);
    }

    response_head =  NetModSocket::decode_response_head(response_head_bs);

    recv_buf = malloc(response_head.content_length);
    if(recv_buf == NULL) {
      err_exit(errno, "NetModSocket::send malloc");
      LoggerFactory::getLogger()->error(errno, "NetModSocket::send malloc");
      exit(1);
    }
    if ( (recv_size = rio_readnb(rio, recv_buf, response_head.content_length) ) !=  response_head.content_length) {
      err_exit(errno, "NetModSocket::send  rio_readnb");
      LoggerFactory::getLogger()->error(errno, "NetModSocket::send  rio_readnb");
      exit(1);
    }

LABEL_ARR_PUSH:
@@ -86,12 +98,70 @@
    ret_arr[i].content_length = recv_size;
  }
  *recv_arr = ret_arr;
  *recv_arr_size = i;

  if(recv_arr_size != NULL) {
    *recv_arr_size = i;
  }
 
  return i;
     
}

// int  pub(char *topic, int topic_size, void *content, int content_size, int port);

int NetModSocket::pub(net_node_t *node_arr, int arrlen, char *topic, int topic_size, void *content, int content_size) {
  int i, n, clientfd;
  char portstr[32];
  net_node_t *node;
  char mapKey[256];
  void *recv_buf;
  int recv_size;
  char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH];
  net_mod_request_head_t request_head;
  net_mod_response_head_t response_head;
  std::map<std::string, rio_t*>::iterator mapIter;
  rio_t *rio;
  for (i = 0; i< arrlen; i++) {

    node = &node_arr[i];
    if(node->host == NULL) {
      // 本地发送
      shmModSocket.pub(topic, topic_size, content, content_size, node->key);
      
    } else {
      sprintf(mapKey, "%s:%d", node->host, node->port);
      if( ( mapIter = connectionMap.find(mapKey)) != connectionMap.end()) {
        rio = mapIter->second;
      } else {
        rio = (rio_t *)malloc(sizeof(rio_t));
        sprintf(portstr, "%d", node->port);
        clientfd = Open_clientfd(node-> host, portstr);
        Rio_readinitb(rio, clientfd);
        connectionMap.insert({mapKey, rio});
      }

      request_head.mod = BUS;
      request_head.key = node->key;
      request_head.content_length = content_size;
      request_head.topic_length = strlen(topic) + 1;
      if(rio_writen(rio->rio_fd, NetModSocket::encode_request_head(request_head), NET_MODE_REQUEST_HEAD_LENGTH) != NET_MODE_REQUEST_HEAD_LENGTH) {
        LoggerFactory::getLogger()->error(errno, "NetModSocket::pub head rio_writen");
        exit(1);

      }

      if(rio_writen(rio->rio_fd, content, content_size) != content_size ) {
        LoggerFactory::getLogger()->error(errno, "NetModSocket::pub rio_writen conent ");
        exit(1);
      }

      if(rio_writen(rio->rio_fd, topic, request_head.topic_length) != request_head.topic_length ) {
        LoggerFactory::getLogger()->error(errno, "NetModSocket::pub rio_writen conent ");
        exit(1);
      }
    }
  }
  return i;
}

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

@@ -112,6 +182,7 @@
  PUT(head, htonl(request.mod));
  PUT(head + 4, htonl(request.key));
  PUT(head + 8, htonl(request.content_length));
  PUT(head + 12, htonl(request.topic_length));
  return head;
}

@@ -121,6 +192,7 @@
  head.mod = ntohl(GET(headbs));
  head.key = ntohl(GET(headbs + 4));
  head.content_length = ntohl(GET(headbs + 8));
  head.topic_length = ntohl(GET(headbs + 12));
  return head;
}