#include "net_mod_socket.h" #include "socket_io.h" #include "net_mod_socket_io.h" std::map NetModSocket::connectionMap; NetModSocket::NetModSocket() { } 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, 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; 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::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; 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) { LoggerFactory::getLogger()->error(errno, "NetModSocket::send head rio_writen"); exit(1); } if(rio_writen(rio->rio_fd, send_buf, send_size) != send_size ) { 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) { 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) { 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) { LoggerFactory::getLogger()->error(errno, "NetModSocket::send rio_readnb"); exit(1); } 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; 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::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) { for(int i =0; i< size; i++) { free(arr[i].content); } free(arr); } // ssize_t recv(void *buf, size_t len) { // return rio_readlineb(&rio, buf, MAXLINE); // } void * NetModSocket::encode_request_head(net_mod_request_head_t & request) { char * head = (char *)malloc(NET_MODE_REQUEST_HEAD_LENGTH); 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; } net_mod_request_head_t NetModSocket::decode_request_head(void *_headbs) { char *headbs = (char *)_headbs; net_mod_request_head_t head; 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; } void * NetModSocket::encode_response_head(net_mod_response_head_t & response) { char * head = (char *)malloc(NET_MODE_RESPONSE_HEAD_LENGTH); PUT(head, htonl(response.content_length)); return head; } net_mod_response_head_t NetModSocket::decode_response_head(void *_headbs) { char *headbs = (char *)_headbs; net_mod_response_head_t head; head.content_length = ntohl(GET(headbs)); return head; }