| | |
| | | #include "socket_io.h" |
| | | #include "net_mod_socket_io.h" |
| | | #include "net_mod_socket.h" |
| | | #include "key_def.h" |
| | | |
| | | static Logger * logger = LoggerFactory::getLogger(); |
| | | |
| | |
| | | sprintf(portstr, "%d", port); |
| | | listenfd = open_listenfd(portstr); |
| | | if(listenfd < 0) { |
| | | LoggerFactory::getLogger()->error(errno, "NetModServerSocket::start"); |
| | | LoggerFactory::getLogger()->error(errno, "NetModServerSocket::start . errno=%d", errno); |
| | | return -1; |
| | | } |
| | | init_pool(listenfd); |
| | |
| | | int NetModServerSocket::process_client(int connfd) { |
| | | net_mod_request_head_t request_head; |
| | | net_mod_response_head_t response_head; |
| | | int ret; |
| | | char request_head_bs[NET_MODE_REQUEST_HEAD_LENGTH]; |
| | | void *recv_buf; |
| | | // char tmp[8196]; |
| | | int recv_size, response_buf_size; |
| | | struct timespec timeout; |
| | | |
| | | if (rio_readn(connfd, request_head_bs, NET_MODE_REQUEST_HEAD_LENGTH) != NET_MODE_REQUEST_HEAD_LENGTH) |
| | | { |
| | |
| | | } |
| | | |
| | | request_head = NetModSocket::decode_request_head(request_head_bs); |
| | | // printf("server received request from host = %s:%d\n", request_head.host, request_head.port); |
| | | |
| | | // printf("server received request from host = %s:%d, key = %d, timeout=%d,\n", |
| | | // request_head.host, request_head.port , request_head.key, request_head.timeout); |
| | | |
| | | if(request_head.content_length > max_buf) { |
| | | |
| | |
| | | return -1; |
| | | } |
| | | |
| | | |
| | | if(request_head.mod == REQ_REP) { |
| | | |
| | | memcpy(response_head.host, request_head.host, NI_MAXHOST); |
| | | response_head.port = request_head.port; |
| | | response_head.key = request_head.key; |
| | | if(shmModSocket.sendandrecv_unsafe(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size) != 0) { |
| | | |
| | | |
| | | if(request_head.timeout > 0) { |
| | | timeout.tv_sec = request_head.timeout / 1000; |
| | | timeout.tv_nsec = (request_head.timeout - timeout.tv_sec * 1000) * 10e6; |
| | | |
| | | // printf(" timeout.tv_sec = %d, timeout.tv_nsec=%ld\n", timeout.tv_sec, timeout.tv_nsec ); |
| | | |
| | | ret = shmModSocket.sendandrecv_unsafe_timeout(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size, &timeout); |
| | | } |
| | | else if(request_head.timeout == 0) { |
| | | ret = shmModSocket.sendandrecv_unsafe_nowait(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size); |
| | | } |
| | | else if(request_head.timeout == -1) { |
| | | ret = shmModSocket.sendandrecv_unsafe(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size); |
| | | } |
| | | |
| | | if( ret != 0) { |
| | | // 没有对应的key |
| | | response_head.code = 1; |
| | | response_head.code = ret; |
| | | response_head.content_length = 0; |
| | | if( rio_writen(connfd, NetModSocket::encode_response_head(response_head), NET_MODE_RESPONSE_HEAD_LENGTH) != NET_MODE_RESPONSE_HEAD_LENGTH ) |
| | | return -1; |
| | |
| | | |
| | | |
| | | } else if(request_head.mod == BUS) { |
| | | // TODO: pub response |
| | | if(request_head.topic_length > max_topic_buf) { |
| | | if( (topic_buf = realloc(topic_buf, request_head.topic_length)) == NULL ) { |
| | | LoggerFactory::getLogger()->error(errno, "NetModServerSocket::process_client realloc topic_buf"); |
| | |
| | | if (rio_readn(connfd, topic_buf, request_head.topic_length) != request_head.topic_length ) { |
| | | return -1; |
| | | } |
| | | LoggerFactory::getLogger()->debug("====server pub %s===\n", buf); |
| | | shmModSocket.pub((char*)topic_buf, request_head.topic_length, buf, request_head.content_length, request_head.key); |
| | | // LoggerFactory::getLogger()->debug("====server pub %s===\n", buf); |
| | | memcpy(response_head.host, request_head.host, NI_MAXHOST); |
| | | response_head.port = request_head.port; |
| | | // response_head.key = request_head.key; |
| | | |
| | | if(request_head.timeout > 0) { |
| | | timeout.tv_sec = request_head.timeout / 1000; |
| | | timeout.tv_nsec = (request_head.timeout - timeout.tv_sec * 1000) * 10e6; |
| | | ret = shmModSocket.pub_timeout((char*)topic_buf, request_head.topic_length, buf, request_head.content_length, BUS_KEY, &timeout); |
| | | } |
| | | else if(request_head.timeout == 0) { |
| | | ret = shmModSocket.pub_nowait((char*)topic_buf, request_head.topic_length, buf, request_head.content_length, BUS_KEY); |
| | | } |
| | | else if(request_head.timeout == -1) { |
| | | ret = shmModSocket.pub((char*)topic_buf, request_head.topic_length, buf, request_head.content_length, BUS_KEY); |
| | | } |
| | | |
| | | response_head.code = ret; |
| | | response_head.content_length = 0; |
| | | if( rio_writen(connfd, NetModSocket::encode_response_head(response_head), NET_MODE_RESPONSE_HEAD_LENGTH) != NET_MODE_RESPONSE_HEAD_LENGTH ) |
| | | return -1; |
| | | |
| | | |
| | | } |
| | | |
| | | return 0; |