#include "net_mod_socket.h" #include "socket_io.h" #include "net_mod_socket_io.h" #include /* See NOTES */ #include #include static Logger *logger = LoggerFactory::getLogger(); static pthread_once_t once = PTHREAD_ONCE_INIT; static pthread_key_t poolKey; NetModSocket::NetModSocket() { if (Signal(SIGPIPE, SIG_IGN) == SIG_ERR) logger->error(errno, "NetModSocket::NetModSocket signal"); } NetModSocket::~NetModSocket() { } int NetModSocket::stop() { return shmModSocket.stop(); } /** * 绑定端口到socket, 如果不绑定则系统自动分配一个 * @return 0 成功， 其他值 失败的错误码 */ int NetModSocket::bind(int key) { return shmModSocket.bind(key); } /** * 强制绑定端口到socket, 适用于程序非正常关闭的情况下，重启程序绑定原来还没释放的key * @return 0 成功， 其他值 失败的错误码 */ int NetModSocket::force_bind( int key) { return shmModSocket.force_bind(key); } int NetModSocket::bind_proc_id(char *buf, int len) { return shmModSocket.bind_proc_id(buf, len); } int NetModSocket::reg(void *pData, int len, void **buf, int *size, const int timeout_ms, int flag) { return shmModSocket.reg(pData, len, buf, size, timeout_ms, flag); } void NetModSocket::int_set(int data) { int_val = data; } void NetModSocket::svr_set(int data) { svr_val = data; } int NetModSocket::int_get(void) { return int_val; } int NetModSocket::svr_get(void) { return svr_val; } // 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) { // return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, -1); // } // int NetModSocket::sendandrecv_timeout(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 msec) { // return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, msec); // } // int NetModSocket::sendandrecv_nowait(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, // net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) { // return _sendandrecv_(node_arr, arrlen, send_buf,send_size, recv_arr, recv_arr_size, 0); // } /* Free thread-specific data buffer */ void NetModSocket::_destroyConnPool_(void *_pool) { NetConnPool *mpool = (NetConnPool *)_pool; delete mpool; } /* One-time key creation function */ 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, _destroyConnPool_); if (ret != 0) { logger->error(ret, "pthread_key_create"); exit(1); } } NetConnPool* NetModSocket::_get_threadlocal_pool() { int ret; NetConnPool *mpool; /* Make first caller allocate key for thread-specific data */ ret = pthread_once(&once, _createConnPoolKey_); if (ret != 0) { LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_once"); exit(1); } mpool = (NetConnPool *)pthread_getspecific(poolKey); if (mpool == NULL) { /* 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"); exit(1); } ret = pthread_setspecific(poolKey, mpool); if (ret != 0) { LoggerFactory::getLogger()->error(errno, "NetModSocket::_sendandrecv_ pthread_setspecific"); exit(1); } } return mpool; } NetConnPool* NetModSocket::_get_pool() { return _get_threadlocal_pool(); } 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, net_mod_err_t ** _err_arr, int *_err_arr_size, int msec ) { int i, recv_size, connfd; net_node_t *node; void *recv_buf = NULL; struct timespec timeout; int ret; int n_req = 0, n_recv_suc = 0, n_err = 0, n_resp =0; net_mod_request_head_t request_head = {}; net_mod_recv_msg_t *ret_arr = (net_mod_recv_msg_t *)calloc(arrlen, sizeof(net_mod_recv_msg_t)); net_mod_err_t *err_arr = (net_mod_err_t *)calloc(arrlen, sizeof(net_mod_err_t)); NetConnPool *mpool = _get_pool(); for (i = 0; i< arrlen; i++) { node = &node_arr[i]; if(node->host == NULL || strcmp(node->host, "") == 0 ) { // 本地发送 if(msec == 0) { ret = shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size, NULL, BUS_NOWAIT_FLAG); } else if(msec > 0){ timeout.tv_sec = msec / 1000; timeout.tv_nsec = (msec - timeout.tv_sec * 1000) * 10e6; ret = shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size, &timeout, BUS_TIMEOUT_FLAG); } else { ret = shmModSocket.sendandrecv(send_buf, send_size, node->key, &recv_buf, &recv_size); } if( ret == 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++; } else { err_arr[n_err].port = 0; err_arr[n_err].key = node->key; err_arr[n_err].code = ret; n_err++; // logger->error("NetModSocket:: _sendandrecv_ to key %d failed. %s", node->key, bus_strerror(ret)); } continue; } if( (connfd = mpool->getConn(node->host, node->port)) < 0 ) { memcpy(err_arr[n_err].host, node->host, sizeof(err_arr[n_err].host)); err_arr[n_err].port = node->port; err_arr[n_err].key = node->key; err_arr[n_err].code = EBUS_NET; n_err++; continue; } request_head.mod = REQ_REP; memcpy(request_head.host, node->host, sizeof(request_head.host)); request_head.port = node->port; request_head.key = node->key; request_head.content_length = send_size; request_head.timeout = msec; // printf("write_request %s:%d\n", request_head.host, request_head.port); if(write_request(connfd, request_head, send_buf, send_size, NULL, 0) != 0) { LoggerFactory::getLogger()->error("write_request failture %s:%d\n", node->host, node->port); memcpy(err_arr[n_err].host, node->host, sizeof(err_arr[n_err].host)); err_arr[n_err].port = node->port; err_arr[n_err].key = node->key; err_arr[n_err].code = EBUS_NET; n_err++; mpool->closeConn( connfd); } else { n_req++; } } // printf(" mpool->maxi = %d\n", mpool->maxi); // printf(" n_req = %d\n", n_req); while(n_resp < n_req) { /* Wait for listening/connected descriptor(s) to become ready */ if( (mpool->nready = poll(mpool->conns, mpool->maxi + 1, msec) ) <= 0) { // wirite_set 和 read_set 在指定时间内都没准备好 break; } // printf("mpool->nready =%d\n", mpool->nready); for (i = 0; (i <= mpool->maxi) && (mpool->nready > 0); i++) { if ( (connfd = mpool->conns[i].fd) > 0 ) { /* If the descriptor is ready, echo a text line from it */ if (mpool->conns[i].revents & POLLIN ) { mpool->nready--; // printf("POLLIN %d\n", connfd); if( (ret = read_response(connfd, ret_arr+n_recv_suc, err_arr + n_err)) == 0) { n_recv_suc++; // 成功收到返回消息，清空读入位 mpool->conns[i].fd = -1; } else if(ret == EBUS_NET) { // 网络错误 logger->error("NetModSocket::_sendandrecv_ read_response key = %d , %s", get_key(), bus_strerror(ret)); mpool->closeConn( connfd); n_err++; // mpool->conns[i].fd = -1; } else { // 代理服务没有转发成功 logger->error("NetModSocket::_sendandrecv_ read_response key = %d , %s", get_key(), bus_strerror(ret)); mpool->conns[i].fd = -1; n_err++; } n_resp++; // printf("read response %d\n", n); } if (mpool->conns[i].revents & POLLOUT ) { // printf("poll POLLOUT %d\n", connfd); } if (mpool->conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) ) { // printf("poll POLLERR %d\n", connfd); mpool->nready--; mpool->closeConn( connfd); } } } } //超时后，关闭超时连接 for (i = 0; i <= mpool->maxi; i++) { if ( (connfd = mpool->conns[i].fd) > 0 ) { // 关闭并清除写入或读取失败的连接 mpool->closeConn( connfd); } } mpool->maxi = -1; if(recv_arr != NULL) { if(n_recv_suc > 0) { *recv_arr = ret_arr; } else { free_recv_msg_arr(ret_arr, n_recv_suc); } } else { free_recv_msg_arr(ret_arr, n_recv_suc); } if(recv_arr_size != NULL) { *recv_arr_size = n_recv_suc; } if(_err_arr != NULL) { if(n_err > 0) { *_err_arr = err_arr; } else { *_err_arr = NULL; *_err_arr_size = 0; free(err_arr); } } else { free(err_arr); } if(_err_arr_size != NULL) { *_err_arr_size = n_err; } return n_recv_suc; } 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, const char *topic, int topic_size, const void *content, int content_size) { return _pub_(node_arr, arrlen, topic, topic_size, content, content_size, -1); } int NetModSocket::pub_nowait(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content, int content_size) { return _pub_(node_arr, arrlen, topic, topic_size, content, content_size, 0); } int NetModSocket::pub_timeout(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content, int content_size, int msec ) { return _pub_(node_arr, arrlen, topic, topic_size, content, content_size, msec); } int NetModSocket::_pub_(net_node_t *node_arr, int arrlen, const char *topic, int topic_size, const void *content, int content_size, int msec) { int i, connfd; net_node_t *node; struct timespec timeout; net_mod_request_head_t request_head; net_mod_recv_msg_t recv_msg; char portstr[32]; int n_req = 0, n_pub_suc = 0, n_resp = 0; int ret; NetConnPool *mpool = _get_pool(); net_mod_err_t err_msg; // 本地发送 if ((node_arr == NULL) || (arrlen == 0)) { if(msec == 0) { ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, NULL, BUS_NOWAIT_FLAG); } else if(msec > 0) { timeout.tv_sec = msec / 1000; timeout.tv_nsec = (msec - timeout.tv_sec * 1000) * 10e6; ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, &timeout, BUS_TIMEOUT_FLAG); } else { ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY); } if(ret == 0 ) { n_pub_suc++; } return n_pub_suc; } for (i = 0; i < arrlen; i++) { node = &node_arr[i]; if(node->host == NULL) { // 本地发送 if(msec == 0) { ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, NULL, BUS_NOWAIT_FLAG); } else if(msec > 0) { timeout.tv_sec = msec / 1000; timeout.tv_nsec = (msec - timeout.tv_sec * 1000) * 10e6; ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY, &timeout, BUS_TIMEOUT_FLAG); } else { ret = shmModSocket.pub(topic, topic_size, content, content_size, SHM_BUS_KEY); } if(ret == 0 ) { n_pub_suc++; } } else { sprintf(portstr, "%d", node->port); if( (connfd = mpool->getConn(node->host, node->port)) < 0 ) { continue; } request_head.mod = BUS; memcpy(request_head.host, node->host, sizeof(request_head.host)); request_head.key = SHM_BUS_KEY; request_head.content_length = content_size; request_head.topic_length = strlen(topic) + 1; request_head.timeout = msec; if(write_request(connfd, request_head, content, content_size, topic, request_head.topic_length) != 0) { LoggerFactory::getLogger()->error(" NetModSocket::_pub_ write_request failture %s:%d\n", node->host, node->port); mpool->closeConn( connfd); } else { n_req++; } } } while(n_resp < n_req) { /* Wait for listening/connected descriptor(s) to become ready */ if( (mpool->nready = poll(mpool->conns, mpool->maxi + 1, msec) ) <= 0) { // wirite_set 和 read_set 在指定时间内都没准备好 break; } // printf("mpool->nready =%d\n", mpool->nready); for (i = 0; (i <= mpool->maxi) && (mpool->nready > 0); i++) { if ( (connfd = mpool->conns[i].fd) > 0 ) { if (mpool->conns[i].revents & POLLIN ) { mpool->nready--; // printf("POLLIN %d\n", connfd); if( (ret = read_response(connfd, &recv_msg, &err_msg)) == 0) { // 成功收到返回消息，清空读入位 mpool->conns[i].fd = -1; n_pub_suc++; } else if(ret == EBUS_NET) { // 网络连接错误 mpool->closeConn( connfd); } else { // 代理服务没有转发成功 mpool->conns[i].fd = -1; } n_resp++; // printf("read response %d\n", n); } if (mpool->conns[i].revents & POLLOUT ) { // printf("poll POLLOUT %d\n", connfd); } if (mpool->conns[i].revents & (POLLRDHUP | POLLHUP | POLLERR) ) { // printf("poll POLLERR %d\n", connfd); mpool->nready--; mpool->conns[i].fd = -1; mpool->closeConn( connfd); } } } } //超时后，关闭超时连接 for (i = 0; i <= mpool->maxi; i++) { if ( (connfd = mpool->conns[i].fd) > 0 ) { // 关闭并清除写入或读取失败的连接 mpool->closeConn( connfd); // mpool->conns[i].fd = -1; } } mpool->maxi = -1; return n_pub_suc; } /** * 发送信息 * @key 发送给谁 * @return 0 成功， 其他值 失败的错误码 */ int NetModSocket::sendto(const void *buf, const int size, const int key, int reset, int data_set){ return shmModSocket.sendto(buf, size, key, 0, 0, reset, data_set); } // 发送信息超时返回。 @sec 秒 ， @nsec 纳秒 int NetModSocket::sendto_timeout(const void *buf, const int size, const int key, int sec, int nsec, int reset, int data_set){ struct timespec timeout = {sec, nsec}; return shmModSocket.sendto(buf, size, key, &timeout, BUS_TIMEOUT_FLAG, reset, data_set); } // 发送信息立刻返回。 int NetModSocket::sendto_nowait(const void *buf, const int size, const int key, int reset, int data_set){ return shmModSocket.sendto(buf, size, key, NULL, BUS_NOWAIT_FLAG, reset, data_set); } /** * 接收信息 * @key 从谁哪里收到的信息 * @return 0 成功， 其他值 失败的错误码 */ int NetModSocket::recvfrom(void **buf, int *size, int *key, int reset, int data_set) { return shmModSocket.recvfrom(buf, size, key, 0, 0, reset, data_set); } // 接受信息超时返回。 @sec 秒 ， @nsec 纳秒 int NetModSocket::recvfrom_timeout(void **buf, int *size, int *key, int sec, int nsec, int reset, int data_set){ struct timespec timeout = {sec, nsec}; return shmModSocket.recvfrom(buf, size, key, &timeout, BUS_TIMEOUT_FLAG, reset, data_set); } int NetModSocket::recvfrom_nowait(void **buf, int *size, int *key, int reset, int data_set){ return shmModSocket.recvfrom(buf, size, key, NULL, BUS_NOWAIT_FLAG, reset, data_set); } int NetModSocket::recvandsend(recvandsend_callback_fn callback, const struct timespec *timeout , int flag, void * user_data ) { return shmModSocket.recvandsend(callback, timeout, flag, user_data); } /** * 发送请求信息并等待接收应答 * @key 发送给谁 * @return 0 成功， 其他值 失败的错误码 */ int NetModSocket::sendandrecv( const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size){ return shmModSocket.sendandrecv(send_buf, send_size, key, recv_buf, recv_size); } // 超时返回。 @sec 秒 ， @nsec 纳秒 int NetModSocket::sendandrecv_timeout( const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size, int sec, int nsec){ struct timespec timeout = {sec, nsec}; return shmModSocket.sendandrecv(send_buf, send_size, key, recv_buf, recv_size, &timeout, BUS_TIMEOUT_FLAG); } int NetModSocket::sendandrecv_nowait( const void *send_buf, const int send_size, const int key, void **recv_buf, int *recv_size) { return shmModSocket.sendandrecv(send_buf, send_size, key, recv_buf, recv_size, NULL, BUS_NOWAIT_FLAG); } /** * 订阅指定主题 * @topic 主题 * @size 主题长度 * @key 总线端口 */ int NetModSocket::sub( void *topic, int size, int key){ return shmModSocket.sub((char *)topic, size, key); } // 超时返回。 @sec 秒 ， @nsec 纳秒 int NetModSocket::sub_timeout( void *topic, int size, int key, int sec, int nsec){ struct timespec timeout = {sec, nsec}; return shmModSocket.sub((char *)topic, size, key, &timeout, BUS_TIMEOUT_FLAG); } int NetModSocket::sub_nowait( void *topic, int size, int key){ return shmModSocket.sub((char *)topic, size, key, NULL, BUS_NOWAIT_FLAG); } /** * 取消订阅指定主题 * @topic 主题 * @size 主题长度 * @key 总线端口 */ int NetModSocket::desub( void *topic, int size, int key){ return shmModSocket.desub((char *)topic, size, key); } // 超时返回。 @sec 秒 ， @nsec 纳秒 int NetModSocket::desub_timeout( void *topic, int size, int key, int sec, int nsec){ struct timespec timeout = {sec, nsec}; return shmModSocket.desub((char *)topic, size, key, &timeout, BUS_TIMEOUT_FLAG); } int NetModSocket::desub_nowait( void *topic, int size, int key){ return shmModSocket.desub((char *)topic, size, key, NULL, BUS_NOWAIT_FLAG); } /** * 发布主题 * @topic 主题 * @content 主题内容 * @key 总线端口 */ int NetModSocket::pub( char *topic, int topic_size, void *content, int content_size, int key){ return shmModSocket.pub(topic, topic_size, content, content_size, key); } // 超时返回。 @sec 秒 ， @nsec 纳秒 int NetModSocket::pub_timeout( char *topic, int topic_size, void *content, int content_size, int key, int sec, int nsec){ struct timespec timeout = {sec, nsec}; return shmModSocket.pub(topic, topic_size, content, content_size, key, &timeout, BUS_TIMEOUT_FLAG); } int NetModSocket::pub_nowait( char *topic, int topic_size, void *content, int content_size, int key){ return shmModSocket.pub(topic, topic_size, content, content_size, key, NULL, BUS_NOWAIT_FLAG); } /** * 获取soket端口号 */ int NetModSocket::get_key() { return shmModSocket.get_key(); } //====================================================================================== int NetModSocket::write_request(int clientfd, net_mod_request_head_t &request_head, const void *content_buf, int content_size, const void *topic_buf, int topic_size) { int buf_size; char *buf; int max_buf_size; if((buf = (char *)malloc(MAXBUF)) == NULL) { LoggerFactory::getLogger()->error(errno, "NetModSocket::NetModSocket malloc"); exit(1); } else { max_buf_size = MAXBUF; } buf_size = NET_MODE_REQUEST_HEAD_LENGTH + content_size + topic_size ; if(max_buf_size < buf_size) { if((buf = (char *)realloc(buf, buf_size)) == NULL) { LoggerFactory::getLogger()->error(errno, "NetModSocket::write_request realloc buf"); exit(1); } else { max_buf_size = buf_size; } } memcpy(buf, NetModSocket::encode_request_head(request_head), NET_MODE_REQUEST_HEAD_LENGTH); memcpy(buf + NET_MODE_REQUEST_HEAD_LENGTH, content_buf, content_size); if(topic_size != 0 ) memcpy(buf + NET_MODE_REQUEST_HEAD_LENGTH + content_size, topic_buf, topic_size); if(rio_writen(clientfd, buf, buf_size) != buf_size ) { LoggerFactory::getLogger()->error(errno, "NetModSocket::write_request rio_writen"); free(buf); return -1; } free(buf); return 0; } /** * @return 0 成功, EBUS_NET 网络错误， 其他值 代理服务没有转发成功。 * */ int NetModSocket::read_response(int connfd, net_mod_recv_msg_t *recv_msg, net_mod_err_t *err_arr) { int recv_size; void *recv_buf; char response_head_bs[NET_MODE_RESPONSE_HEAD_LENGTH]; net_mod_response_head_t response_head; if ( rio_readn(connfd, response_head_bs, NET_MODE_RESPONSE_HEAD_LENGTH) != NET_MODE_RESPONSE_HEAD_LENGTH) { LoggerFactory::getLogger()->error(errno, "NetModSocket::read_response rio_readnb response_head"); memcpy(err_arr->host, "unkown", sizeof(err_arr->host)); err_arr->port = 0; err_arr->key = 0; err_arr->code = EBUS_NET; return EBUS_NET; } response_head = NetModSocket::decode_response_head(response_head_bs); // printf(">>>> read_response %s\n", response_head.host); if(response_head.code != 0) { // 代理服务没能成功发送给对应的key memcpy(err_arr->host, response_head.host, sizeof(err_arr->host)); err_arr->port = response_head.port; err_arr->key = response_head.key; err_arr->code = response_head.code; return response_head.code; } recv_buf = malloc(response_head.content_length); if(recv_buf == NULL) { LoggerFactory::getLogger()->error(errno, "NetModSocket::send malloc recv_buf"); exit(1); } if ( (recv_size = rio_readn(connfd, recv_buf, response_head.content_length) ) != response_head.content_length) { memcpy(err_arr->host, response_head.host, sizeof(err_arr->host)); err_arr->port = response_head.port; err_arr->key = response_head.key; err_arr->code = EBUS_NET; LoggerFactory::getLogger()->error(errno, "NetModSocket::read_response rio_readnb recv_buf"); //网络错误 return EBUS_NET; } strcpy( recv_msg->host, response_head.host); recv_msg->port = response_head.port; recv_msg->key = response_head.key; recv_msg->content = recv_buf; recv_msg->content_length = recv_size; return 0; } /** uint32_t mod; char host[NI_MAXHOST]; uint32_t port; uint32_t key; uint32_t content_length; uint32_t topic_length; */ void * NetModSocket::encode_request_head(net_mod_request_head_t & head) { void * headbs = malloc(NET_MODE_REQUEST_HEAD_LENGTH); char *tmp_ptr = (char *)headbs; PUT(tmp_ptr, htonl(head.mod)); tmp_ptr += 4; memcpy(tmp_ptr, head.host, sizeof(head.host)); tmp_ptr += sizeof(head.host); PUT(tmp_ptr, htonl(head.port)); tmp_ptr += 4; PUT(tmp_ptr, htonl(head.key)); tmp_ptr += 4; PUT(tmp_ptr, htonl(head.content_length)); tmp_ptr += 4; PUT(tmp_ptr, htonl(head.topic_length)); tmp_ptr += 4; PUT_INT32(tmp_ptr, htonl(head.timeout)); return headbs; } net_mod_request_head_t NetModSocket::decode_request_head(void *headbs) { char *tmp_ptr = (char *)headbs; net_mod_request_head_t head; head.mod = ntohl(GET(tmp_ptr)); tmp_ptr += sizeof(uint32_t); memcpy(head.host, tmp_ptr, sizeof(head.host)); tmp_ptr += sizeof(head.host); head.port = ntohl(GET(tmp_ptr)); tmp_ptr += sizeof(uint32_t); head.key = ntohl(GET(tmp_ptr)); tmp_ptr += sizeof(uint32_t); head.content_length = ntohl(GET(tmp_ptr)); tmp_ptr += sizeof(uint32_t); head.topic_length = ntohl(GET(tmp_ptr)); tmp_ptr += sizeof(uint32_t); head.timeout = ntohl(GET_INT32(tmp_ptr)); return head; } /** char host[NI_MAXHOST]; uint32_t port; uint32_t key; uint32_t content_length; uint32_t code; */ void * NetModSocket::encode_response_head(net_mod_response_head_t & response) { void * headbs = malloc(NET_MODE_RESPONSE_HEAD_LENGTH); char *tmp_ptr = (char *)headbs; memcpy(tmp_ptr, response.host, NI_MAXHOST); tmp_ptr += NI_MAXHOST; PUT(tmp_ptr, htonl(response.port)); tmp_ptr += 4; PUT(tmp_ptr , htonl(response.key)); tmp_ptr += 4; PUT(tmp_ptr, htonl(response.content_length)); tmp_ptr += 4; PUT(tmp_ptr, htonl(response.code)); return headbs; } net_mod_response_head_t NetModSocket::decode_response_head(void *headbs) { char *tmp_ptr = (char *)headbs; net_mod_response_head_t head; memcpy(head.host, tmp_ptr, NI_MAXHOST); tmp_ptr += NI_MAXHOST; head.port = ntohl(GET(tmp_ptr)); tmp_ptr += 4; head.key = ntohl(GET(tmp_ptr)); tmp_ptr += 4; head.content_length = ntohl(GET(tmp_ptr)); tmp_ptr += 4; head.code = ntohl(GET(tmp_ptr)); return head; }