异步Windows Socket包装，包括TCP与UDP，可处理粘包

头文件

struct IPAddress
{
	union {
		struct IPV4Address
		{
			ULONG host;
			USHORT port;
		}address;
		UINT64 _64;
	};

	static IPAddress Zero;

	ULONG& GetHost(){return address.host;}
	USHORT& GetPort(){return address.port;}
	ULONG GetHost()const{return address.host;}
	USHORT GetPort()const{return address.port;}
	string ToString()
	{
		in_addr addr;
		addr.S_un.S_addr = address.host;
		stringstream ss;
		ss << inet_ntoa(addr) << ":" << address.port;
		return ss.str();
	}

	IPAddress(ULONG _host = 0, USHORT _port = 0)
	{
		_64 = 0;
		address.host = _host;
		address.port = _port;
	}

	bool operator < (const IPAddress& _right) const
	{
		return _64 < _right._64;
	}

	bool operator != (const IPAddress& _right) const
	{
		return _64 != _right._64;
	}

	bool operator == (const IPAddress& _right) const
	{
		return _64 == _right._64;
	}
};

class Socket
{
public:

	Socket(int proto = SOCK_STREAM);
	Socket(SOCKET s);
	virtual ~Socket();

	u_long ReceiveLen();

	void Close();

protected:
	friend class SocketSelect;
	SOCKET s_;

	int* refCounter_;
	static int  nofSockets_;

	static void Start();
	static void End();
};

class TcpClient : public Socket
{
protected:
	friend class TcpServer;
	bool connected;

	TcpClient();
	TcpClient(SOCKET s);
public:
	int SendBytes(const std::string&);
	int Send(void* buf, int len);
	int Receive(char* buf, int len);
	bool IsConnected(){return connected;}
	string ReceiveBytes();
	TcpClient(const TcpClient&);
	TcpClient& operator=(TcpClient&);
	TcpClient(const std::string& host, int port);
	TcpClient(unsigned long host, int port);
};

class UdpSocket : public Socket
{
public:
	UdpSocket(unsigned long host, int port);
	int SendBytes(const std::string&, unsigned long host, int port);
	int Send(void* buf, int len, unsigned long host, int port);
	int Send(void* buf, int len, const IPAddress& addr);
	int Receive(char* buf, int len, unsigned long& host, int& port);
	int ReadEventSelect(HANDLE _event);
};

class SocketSelect
{
	static TIMEVAL ms_tval;
	fd_set fds_;
	Socket* sock_;
public:
	SocketSelect(Socket * sock);

	bool Readable();
	bool Writeable();
};

class TcpServer : public Socket
{
	auto_ptr<SocketSelect> select;
	int Init(int port, int connections);
	
public:
	unsigned short listenPort;
	TcpServer(int port, int connections);

	TcpClient* Accept(u_long& host, USHORT& port);
};

class TcpPacketSeparator
{
	int cur_;
	int len_;
	char* data_;
public:
	TcpPacketSeparator(int maxdatalen)
	{
		len_ = maxdatalen;
		data_ = new char[maxdatalen];
		cur_ = 0;
	}

	~TcpPacketSeparator()
	{
		delete data_;
	}

	bool Process(TcpClient* s)
	{
		u_long len = 0;
		int recv_len_ = 0;

		if( cur_ >= 4 && cur_ >= *(unsigned short*)data_ )
		{
			return true;
		}

		while( (recv_len_ = s->Receive(data_ + cur_, len_ - cur_) ) > 0 )
		{
			cur_ += recv_len_;

			if( cur_ >= *(unsigned short*)data_ )
			{
				return true;
			}
		}
		
		if( recv_len_ == SOCKET_ERROR && WSAGetLastError() != WSAEWOULDBLOCK )
		{
			s->Close();
			return false;
		}

		return false;
	}

	char* GetDataPtr()
	{
		return data_;
	}

	void Done()
	{
		u_short pklen_ = *(unsigned short*)data_;
		if( cur_ > pklen_ )
		{
			memmove_s(data_, len_, data_ + pklen_, cur_ - pklen_);
			cur_ -= pklen_;
		}
		else
		{
			cur_ = 0;
		}
	}
};

CPP

#include <iostream>

using namespace std;

int Socket::nofSockets_= 0;
IPAddress IPAddress::Zero;

void Socket::Start()
{
	if (!nofSockets_)
	{
		WSADATA info;
		if (WSAStartup(MAKEWORD(2,0), &info))
		{
			throw "Could not start WSA";
		}
	}
	++nofSockets_;
}

void Socket::End()
{
	WSACleanup();
}

Socket::Socket(int proto) : s_(0)
{
	Start();
	// UDP: use SOCK_DGRAM instead of SOCK_STREAM
	s_ = socket(AF_INET,proto,proto==SOCK_DGRAM?IPPROTO_UDP:0);

	if (s_ == INVALID_SOCKET)
	{
		throw "INVALID_SOCKET";
	}

	refCounter_ = new int(1);
}

Socket::Socket(SOCKET s)
{
	Start();
	s_ = s;
	refCounter_ = new int(1);
}

Socket::~Socket()
{
	if (! --(*refCounter_))
	{
		Close();
		delete refCounter_;
	}

	--nofSockets_;
	if (!nofSockets_) End();
}

void Socket::Close()
{
	closesocket(s_);
}

u_long Socket::ReceiveLen()
{
	u_long arg = 0;
	if (ioctlsocket(s_, FIONREAD, &arg) != 0)
		return 0;

	return arg;
}

TcpClient::TcpClient() : Socket(SOCK_STREAM)
{
}

TcpClient::TcpClient(const TcpClient& o)
{
	Start();
	refCounter_=o.refCounter_;
	(*refCounter_)++;
	s_         =o.s_;
	nofSockets_++;
}

TcpClient& TcpClient::operator=(TcpClient& o)
{
	refCounter_=o.refCounter_;
	(*refCounter_)++;
	s_         =o.s_;
	nofSockets_++;

	return *this;
}

int TcpClient::Receive(char* buf, int len)
{
	return recv (s_, buf, len, 0);
}

std::string TcpClient::ReceiveBytes()
{
	std::string ret;
	char buf[1024];

	while (1)
	{
		u_long arg = 0;
		if (ioctlsocket(s_, FIONREAD, &arg) != 0)
			break;

		if (arg == 0)
			break;

		if (arg > 1024) arg = 1024;

		int rv = recv (s_, buf, arg, 0);
		if (rv <= 0) break;

		std::string t;

		t.assign (buf, rv);
		ret += t;
	}

	return ret;
}

int TcpClient::SendBytes(const std::string& s)
{
	return send(s_,s.c_str(),(int)s.length(),0);
}

int TcpClient::Send(void* buf, int len)
{
	return send(s_,(const char*)buf,len,0);
}

TcpClient::TcpClient(const std::string& host, int port) : connected(false)
{
	hostent *he;
	if ((he = gethostbyname(host.c_str())) == 0)
	{
		connected = false;
	}

	sockaddr_in addr;
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	addr.sin_addr = *((in_addr *)he->h_addr);
	memset(&(addr.sin_zero), 0, 8); 

	if (::connect(s_, (sockaddr *) &addr, sizeof(sockaddr)))
	{
		connected = false;
	}
	else
	{
		connected = true;
	}

	u_long arg = 1;
	int ret = ioctlsocket(s_, FIONBIO, &arg);
}

TcpClient::TcpClient(unsigned long host, int port) : connected(false)
{
	sockaddr_in addr;
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	addr.sin_addr.S_un.S_addr = host;
	memset(&(addr.sin_zero), 0, 8);

	if (::connect(s_, (sockaddr *) &addr, sizeof(sockaddr)))
	{
		connected = false;
	}
	else
	{
		connected = true;
	}

	u_long arg = 1;
	int ret = ioctlsocket(s_, FIONBIO, &arg);
}

TcpClient::TcpClient(SOCKET s):Socket(s)
{
}

// 用来处理10054 UDP错误
// MS Transport Provider IOCTL to control
// reporting PORT_UNREACHABLE messages
// on UDP sockets via recv/WSARecv/etc.
// Path TRUE in input buffer to enable (default if supported),
// FALSE to disable.
#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR,12)

UdpSocket::UdpSocket(unsigned long host, int port):Socket(SOCK_DGRAM)
{
	BOOL bNewBehavior = FALSE;
	DWORD dwBytesReturned = 0;
	// 用来处理10054 UDP错误
	//WSAIoctl(s_, SIO_UDP_CONNRESET, &bNewBehavior, sizeof bNewBehavior, NULL, 0, &dwBytesReturned, NULL, NULL);

	sockaddr_in sa;

	memset(&sa, 0, sizeof(sa));
	
	if( host == 0 )
	{
		char name[128];
		gethostname(name, 128);
		hostent *pHost = gethostbyname(name);
		if( pHost )
			host = **(DWORD**)(pHost-> h_addr_list);
	}

	sa.sin_family = PF_INET;
	sa.sin_addr.S_un.S_addr = host;
	sa.sin_port = htons(port);

	while(true)
	{
		if (bind(s_, (sockaddr *)&sa, sizeof(sockaddr_in)) == SOCKET_ERROR)
		{
			sa.sin_port = htons(port++);
			if( port > 60000 )
			{
				closesocket(s_);
				break;
			}
			continue;
		}
		
		u_long arg = 1;
		//int ret = ioctlsocket(s_, FIONBIO, &arg);

		break;
	}
}

int UdpSocket::SendBytes(const std::string& buffer, unsigned long host, int port)
{
	sockaddr_in destaddr;
	destaddr.sin_family = AF_INET;
	destaddr.sin_port = htons(port);
	destaddr.sin_addr.s_addr = host;

	return sendto(s_, buffer.c_str(), int(buffer.length()), 0, (SOCKADDR *) &destaddr, sizeof(destaddr));
}

int UdpSocket::Send(void* buf, int len, unsigned long host, int port)
{
	//return 0;
	sockaddr_in destaddr;
	destaddr.sin_family = AF_INET;
	destaddr.sin_port = htons(port);
	destaddr.sin_addr.s_addr = host;

	return sendto(s_, (char*)buf, len, 0, (SOCKADDR *) &destaddr, sizeof(destaddr));
}

int UdpSocket::Send(void* buf, int len, const IPAddress& addr)
{
	return Send(buf, len, addr.GetHost(), addr.GetPort());
}

int UdpSocket::Receive(char* buf, int len, unsigned long& host, int& port)
{
	sockaddr_in senderaddr;
	int destlen = sizeof(senderaddr);
	int ret = recvfrom(s_, buf, len, 0, (SOCKADDR *) &senderaddr, &destlen);
	host = senderaddr.sin_addr.s_addr;
	port = (int)ntohs(senderaddr.sin_port);
	return ret;
}

int UdpSocket::ReadEventSelect(HANDLE _event)
{
	return WSAEventSelect(s_, _event, FD_READ);
}

TcpServer::TcpServer(int port, int connections):listenPort(port)
{
	while( 0 != Init(listenPort, connections) )
	{
		listenPort++;
		if( listenPort > 60000 )
		{
			closesocket(s_);
			return;;
		}
	}
	select.reset(new SocketSelect(this));
}

int TcpServer::Init(int port, int connections)
{
	sockaddr_in sa;

	memset(&sa, 0, sizeof(sa));

	sa.sin_family = PF_INET;             
	sa.sin_port = htons(port);

	if (bind(s_, (sockaddr *)&sa, sizeof(sockaddr_in)) == SOCKET_ERROR)
	{
		return WSAGetLastError();
	}
	
	u_long arg = 1;
	int ret = ioctlsocket(s_, FIONBIO, &arg);

	listen(s_, connections);
	return 0;
}

TcpClient* TcpServer::Accept(u_long& host, USHORT& port)
{
	sockaddr_in sa;
	memset(&sa, 0, sizeof(sa));
	int destlen = sizeof(sa);

	SOCKET new_sock = accept(s_, (sockaddr *)&sa, &destlen);
	if (new_sock == INVALID_SOCKET)
	{
		int rc = WSAGetLastError();
		if(rc==WSAEWOULDBLOCK)
		{
			return 0; // non-blocking call, no request pending
		}
		else
		{
			return 0;
		}
	}
	
	u_long arg = 1;
	int ret = ioctlsocket(s_, FIONBIO, &arg);

	host = sa.sin_addr.S_un.S_addr;
	port = ntohs(sa.sin_port);

	TcpClient* r = new TcpClient(new_sock);
	return r;
}

TIMEVAL SocketSelect::ms_tval = {0, 1};

SocketSelect::SocketSelect(Socket * sock):sock_(sock)
{
}

bool SocketSelect::Readable()
{
	FD_ZERO(&fds_);
	FD_SET(const_cast<Socket*>(sock_)->s_,&fds_);

	if (select (0, &fds_, (fd_set*) 0, (fd_set*) 0, &ms_tval) == SOCKET_ERROR) 
		return true;

	if (FD_ISSET(sock_->s_,&fds_)) return true;
	return false;
}

bool SocketSelect::Writeable()
{
	FD_ZERO(&fds_);
	FD_SET(const_cast<Socket*>(sock_)->s_,&fds_);

	if (select (0, (fd_set*) 0, &fds_, (fd_set*) 0, &ms_tval) == SOCKET_ERROR) 
		return true;

	if (FD_ISSET(sock_->s_,&fds_)) return true;
	return false;
}