version

wangzhengquan

2020-10-12 d26e81c4213dfb04391c7b6692f243aede2e6895

 src/socket/net_mod_socket.c

@@ -2,24 +2,95 @@
#include "socket_io.h"
#include "net_mod_socket_io.h"

NetModSocket::NetModSocket(const char *host, int port) 

std::map<std::string, rio_t *> NetModSocket::connectionMap;

NetModSocket::NetModSocket() 
{
   char portstr[32];
   sprintf(portstr, "%d", port);
  clientfd = Open_clientfd(host, portstr);
  Rio_readinitb(&rio, clientfd);
		
}


ssize_t NetModSocket::send(void *buf, size_t size) {
  int n = rio_writen(clientfd, buf, size);
  rio_writen(clientfd, PKG_SEP, strlen(PKG_SEP));
int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** resp_arr, int *resp_arr_size) {

  char resp[MAXLINE];
  int ss;
  ss = rio_readpkgb(&rio, resp, MAXLINE);
  puts(resp);
  return n;
  int i, n, clientfd;
  char portstr[32];
  net_node_t *node;
  char mapKey[256];
  void *recv_buf;
  int recv_size;
  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;
  net_mod_recv_msg_t *ret_arr = (net_mod_recv_msg_t *)calloc(arrlen, sizeof(net_mod_recv_msg_t));
  
  for (i = 0; i< arrlen; i++) {

    node = &node_arr[i];
    if(node->host == NULL) {
      // 本地发送
      shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size);
      goto LABEL_ARR_PUSH;
    }
    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 = REQ_REP;
    request_head.key = node->key;
    request_head.content_length = send_size;
    if( (n = rio_writen(rio->rio_fd, &request_head, sizeof(request_head))) != sizeof(request_head)) {
      err_exit(errno, "NetModSocket::send head rio_writen");

    }

    if( (n = rio_writen(rio->rio_fd, send_buf, send_size)) != send_size ) {
       err_exit(errno, "NetModSocket::send conent rio_writen");
    }

    if ((n = rio_readnb(rio, &response_head, sizeof(response_head))) !=  sizeof(response_head)) {
      err_exit(errno, "NetModSocket::send  rio_readnb");
    }

    recv_buf = malloc(response_head.content_length);
    if(recv_buf == NULL) {
      err_exit(errno, "NetModSocket::send malloc");
    }
    if ( (recv_size = rio_readnb(rio, recv_buf, response_head.content_length) ) !=  response_head.content_length) {
      err_exit(errno, "NetModSocket::send  rio_readnb");
    }

LABEL_ARR_PUSH:
    strcpy( ret_arr[i].host, node->host);
    ret_arr[i].port = node->port;
    ret_arr[i].key = node->key;
    ret_arr[i].content = recv_buf;
    ret_arr[i].content_length = recv_size;
  }
  *resp_arr = ret_arr;
  *resp_arr_size = i;

  return i;
     
}


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);
  }

}

// ssize_t recv(void *buf, size_t len) {
@@ -29,6 +100,6 @@
// }

NetModSocket::~NetModSocket() {
   Close(clientfd);
  
}