req_rep

wangzhengquan

2020-10-12 575f2e424a17737111786227103a428fb5c20396

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
#include "net_mod_socket.h"
#include "socket_io.h"
#include "net_mod_socket_io.h"
 
 
std::map<std::string, rio_t *> NetModSocket::connectionMap;
 
NetModSocket::NetModSocket() 
{
        
}
 
 
int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {
 
  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;
  }
  *recv_arr = ret_arr;
  *recv_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) {
 
//     return rio_readlineb(&rio, buf, MAXLINE);
  
// }
 
NetModSocket::~NetModSocket() {
  
}