Completionportmode

#include "connection.h" #include <stdio.h> #pragma comment(lib, "ws2_32.lib") HANDLE Connection :: hCompletionPort = NULL; Connection :: Connection() { ZeroMemory(this, sizeof(Connection)); handle = NULL; status = SOCK_NULL; } Connection :: Connection(int type, Address* pAddr, SOCK_CALLBACK fCallback) { int temp; int ret; struct sockaddr_in addr; ZeroMemory(this, sizeof(Connection)); status = SOCK_NULL; if (type == TYPE_UDP) { socktype = SOCK_DGRAM; } else { socktype = SOCK_STREAM; } handle = socket(AF_INET, socktype, 0); if (handle != INVALID_SOCKET) { temp = 1; callback = fCallback; ret = setsockopt(handle, SOL_SOCKET,SO_REUSEADDR, (char*)&temp, sizeof(temp)); if (ret == SOCKET_ERROR) { status = SOCK_ERROR; } else { temp = 1; setsockopt(handle, SOL_SOCKET, TCP_NODELAY, (char*)&temp, sizeof(temp)); if (pAddr == NULL) { addr.sin_family = AF_INET; addr.sin_addr.s_addr = htonl(INADDR_ANY); addr.sin_port = 0; ret = bind(handle, (sockaddr*)&addr, sizeof(addr)); } else { ret = bind(handle, (sockaddr*)pAddr->GetRawAddress(), sizeof(sockaddr_in)); } if (ret != 0) { status = SOCK_ERROR; } } } if (status != SOCK_ERROR && socktype == SOCK_DGRAM) { status = SOCK_CONN; AddToCompletionPort(); } } Connection* Connection :: Create(int type, Address* pAddr, SOCK_CALLBACK callback) { return new Connection(type, pAddr, callback); } Connection :: ~Connection() { if (status == SOCK_CONN) { Close(); } } void* Connection :: GetExtData() { return extdata; } void* Connection :: SetExtData(void* data) { void* p = extdata; extdata = data; return p; } int Connection :: GetRemoteAddress(Address* pAddr) { *pAddr = remote; return 0; } int Connection :: GetLocalAddress(Address* pAddr) { struct sockaddr_in addr; int length = sizeof(addr); if ( getsockname(handle, (sockaddr*)&addr, &length) == SOCKET_ERROR) { return RESULT_ERROR; } pAddr->SetRawAddress(&addr); return 0; } int Connection :: AddToCompletionPort() { int ret; io_recv_data = (IO_RECV_DATA*)GlobalAlloc(GPTR, sizeof(IO_RECV_DATA)); io_recv_data->type.type = SOCK_RECV; io_recv_data->count = 0; status = SOCK_CONN; ret = (int)CreateIoCompletionPort((HANDLE)handle, hCompletionPort, (DWORD)this, 0); if (ret == NULL) { delete io_recv_data; ret = RESULT_ERROR; } else { PostQueuedCompletionStatus(hCompletionPort, -1, (DWORD)this, NULL); ret = RESULT_OK; } return ret; } int Connection :: Connect(Address* addr) { int ret; if (socktype == SOCK_STREAM) { ret = connect(handle, (sockaddr*)addr->GetRawAddress(), sizeof(struct sockaddr_in)); if (ret == 0) { remote = *addr; return AddToCompletionPort(); } else { return RESULT_ERROR; } } else { ret = RESULT_OK; remote = *addr; } return ret; } int Connection :: IsTCPType() { return (socktype == SOCK_STREAM); } int Connection :: IsUDPType() { return (socktype == SOCK_DGRAM); } int Connection :: Listen(int backlog) { int ret; ret = listen(handle, backlog); if (ret == 0) { return RESULT_OK; } return RESULT_ERROR; } Connection* Connection :: Accept() { struct sockaddr_in addr; int n = sizeof(sockaddr); Connection* conn = new Connection(); conn->handle = accept(handle, (struct sockaddr*)&addr, &n); if (conn->handle != INVALID_SOCKET) { conn->socktype = SOCK_STREAM; conn->status = SOCK_CONN; conn->callback = callback; conn->remote.SetRawAddress(&addr); n = 1; setsockopt(conn->handle, SOL_SOCKET, TCP_NODELAY, (char*)&n, sizeof(n)); if (conn->AddToCompletionPort() == RESULT_ERROR) { delete conn; conn = NULL; } } else { delete conn; conn = NULL; } return conn; } int Connection :: SendData(char* data, int nLength, Address* addr) { int ret = SOCKET_ERROR; DWORD count = 0; IO_SEND_DATA* send_data = NULL; WSABUF buff[1]; if (socktype == SOCK_DGRAM) { buff[0].buf = data; buff[0].len = nLength; if (addr == NULL) { addr = &remote; } send_data = (IO_SEND_DATA*)GlobalAlloc(GPTR, sizeof(IO_SEND_DATA) + nLength); send_data->type.type = SOCK_SEND; memcpy(send_data->buffer, data, nLength); ZeroMemory(&send_data->type.Overlapped, sizeof(OVERLAPPED)); ret = WSASendTo(handle, buff, 1, &count, 0, (sockaddr*)addr->GetRawAddress(), sizeof(struct sockaddr_in), &send_data->type.Overlapped, NULL); } else { send_data = (IO_SEND_DATA*)GlobalAlloc(GPTR, sizeof(IO_SEND_DATA) + nLength); send_data->type.type = SOCK_SEND; buff[0].buf = send_data->buffer; buff[0].len = nLength + 4; *(int*)send_data->buffer = nLength; memcpy(send_data->buffer + 4, data, nLength); ZeroMemory(&send_data->type.Overlapped, sizeof(OVERLAPPED)); ret = WSASend(handle, buff, 1, &count, 0, &send_data->type.Overlapped, NULL); } if (ret == SOCKET_ERROR && WSAGetLastError() != ERROR_IO_PENDING) { return RESULT_ERROR; } return RESULT_OK; } int Connection :: RecvData(char* data, int nLength, Address* addr) { int length = 0; if (socktype == SOCK_DGRAM) { length = io_recv_data->count; if (nLength < length) { length = RESULT_ERROR; } else { CopyMemory(data, io_recv_data->buffer, length); } io_recv_data->count = 0; } else { length = *(int*)(io_recv_data->buffer + io_recv_data->offset); if (nLength < length) { length = RESULT_ERROR; } else { CopyMemory(data, io_recv_data->buffer + io_recv_data->offset + 4, length); io_recv_data->offset += 4 + length; io_recv_data->count -= 4 + length; if (io_recv_data->count == 0) { io_recv_data->offset = 0; } } } return length; } int Connection :: Close() { status = SOCK_CLOSE; return closesocket(handle); } // 再次提交一个读数据请求 int Connection :: PostRequest() { int ret; DWORD flag = 0; WSABUF buff[1]; DWORD RecvBytes; buff->buf = io_recv_data->buffer + io_recv_data->count; buff->len = BUFFER_SIZE - io_recv_data->count; ZeroMemory(&io_recv_data->type.Overlapped, sizeof(io_recv_data->type.Overlapped)); if (socktype == SOCK_DGRAM) { int temp = sizeof(sockaddr); ret = WSARecvFrom(handle, buff, 1, &RecvBytes, &flag, (sockaddr*)&io_recv_data->notify.addr, &temp, &io_recv_data->type.Overlapped, NULL); } else { ret = WSARecv(handle, buff, 1, &RecvBytes, &flag, &io_recv_data->type.Overlapped, NULL); } if (ret == SOCKET_ERROR && WSAGetLastError() == ERROR_IO_PENDING) { ret = RESULT_OK; } else { ret = RESULT_ERROR; } if (ret == RESULT_ERROR) { printf("Post Request error!\n"); } else { printf("Post Request OK!\n"); } return ret; } int Connection :: WorkerThread(void* param) { DWORD BytesTransferred = 0; IO_DATA* data = NULL; Connection* conn = NULL; while (TRUE) { if (GetQueuedCompletionStatus(hCompletionPort, &BytesTransferred, (LPDWORD)&conn, (LPOVERLAPPED *)&d