nettcpmodule.cc

五行MMORPG引擎系统V1.0

// SimpleTCPIOModule.cpp: implementation of the NetTCPModule class.
//
//////////////////////////////////////////////////////////////////////
//#include <platforms.h>




//#include <winsock.h>
//#include "server/strhandle.h"
#include "platform/platform.h"
#include "platform/event.h"
#include "server/net/NetTCPModule.h"
#include "core/dataChunker.h"

namespace CS
{

#define	RECVTRY_TIMES	3


typedef struct _tagNetTCPSocket 
{
	NetSocket		sock;
	NetAddress		sockAddr;//rmt;
} NETTCPSOCKET;


static FreeListChunker<NETTCPSOCKET>	gs_netTCPSocketPool;


//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////
//CS_IOHANDLER NetTCPModule::m_ioHandler;
static NetTCPModule gs_netTCPModule;

NetTCPModule::NetTCPModule()
{
	m_ioHandler.pGetError	= GetError;
	m_ioHandler.pClose		= Close;
	m_ioHandler.pConnect	= Connect;
	m_ioHandler.pFree		= Free;
	m_ioHandler.pListen	= Listen;
	m_ioHandler.pCheckAccept	= CheckAccept;
	m_ioHandler.pCheckSend	= CheckSend;
	m_ioHandler.pCheckRecv	= CheckRecv;
	m_ioHandler.pAccept	= Accept;
	m_ioHandler.pQuery		= Query;
	m_ioHandler.pRecvEx	= RecvEx;
	m_ioHandler.pSendEx	= SendEx;
	m_ioHandler.pRecv		= Recv;
	m_ioHandler.pSend		= Send;
	m_ioHandler.pInsert	= Insert;
	m_ioHandler.pRemove	= Remove;
	m_ioHandler.pGetConnectAddr	= GetConnectAddr;
	m_ioHandler.pGetNetAddress	= GetNetAddress;
}

NetTCPModule::~NetTCPModule()
{
}


CS_IOHANDLER* NetTCPModule::GetIOHandler()
{
	return &gs_netTCPModule.m_ioHandler;
	//return &m_ioHandler;
}


INT	RPGAPI NetTCPModule::GetError(void *data)
{
	data;
	return Net::getLastError();
}


int RPGAPI NetTCPModule::Insert()
{
	return Net::init()? Net::OK : Net::NetError;
}

int RPGAPI NetTCPModule::Remove()
{
	Net::shutdown();
	return Net::OK;
}




void * RPGAPI NetTCPModule::Listen(char *svTarget)
{
	//格式
	//	80					端口，默认本地地址
	// 127.0.0.1:80	ip:port
	if(svTarget == NULL) 
		return NULL;

	NETTCPSOCKET	*tcps;
	NetAddress		netAddr;

	Net::stringToAddress(svTarget, &netAddr);
	
	///////////////////////////////////////////////////
	//生成监听socket
	// Create listener socket
	NetSocket sock;
	sock	= Net::openSocket();

	if(sock == InvalidSocket)
		return NULL;
	
	
	//Net::setOption(sock, Net::N_TCP_NODELAY, TRUE, Net::N_IPPROTO_TCP);
	//Net::setOption(sock, Net::N_SO_DONTLINGER, TRUE);


	///////////////////////////////////////////////////
	//绑定socket、启动socket监听
	// Bind socket and listen
	
	if(Net::bind(sock, &netAddr) != Net::NoError) 
	{
		Net::closeSocket(sock);
		return NULL;
	}

	if(Net::listen(sock,Net::N_SOMAXCONN) != Net::NoError) 
	{
		Net::closeSocket(sock);
		return NULL;
	}
	

	//AssertWarn(0,"内存池管理 ");
	// Allocate state structure
	tcps = gs_netTCPSocketPool.alloc();
	if(tcps==NULL) 
	{
		Net::closeSocket(sock);
		return NULL;
	}

	// Fill in state structure
	tcps->sock		= sock;
	tcps->sockAddr	= netAddr;
	//dMemcpy(&(tcps->sockAddr), &netAddr, sizeof(netAddr));
	
	return tcps;
}



BOOL RPGAPI NetTCPModule::CheckAccept(void *data)
{
	// Check to see if this is a listening socket
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)data;

	return Net::checkAccept(tcps->sock,0,0) == Net::NoError;

	//检测是否有连接请求
	// Check for connection
	//fd_set	rdfds;
	//TIMEVAL	tm = {0,0};

	//FD_ZERO(&rdfds);
	//FD_SET(tcps->sock,&rdfds);

	//return select(0,&rdfds,NULL,NULL,&tm) > 0;
}


void * RPGAPI NetTCPModule::Accept(void *data, char *svAddr, int nMaxLen)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)data;

	//接受新的socket请求
	// Accept socket
	//SOCKADDR_IN sockAddr;
	//int			len = sizeof(SOCKADDR_IN);
	NetAddress		addrNew;
	NetSocket		sockNew;

	sockNew = Net::accept(tcps->sock, &addrNew);
	if(sockNew == InvalidSocket)
		return NULL;

	//Net::setOption(sockNew, Net::N_TCP_NODELAY, TRUE, Net::N_IPPROTO_TCP);
	//Net::setOption(sockNew, Net::N_SO_DONTLINGER, TRUE);
	

	//新建IO Socket对象
	NETTCPSOCKET *ioSockNew = gs_netTCPSocketPool.alloc();//(NETTCPSOCKET*)dMalloc(sizeof(NETTCPSOCKET));
	if(ioSockNew == NULL) 
	{
		Net::closeSocket(sockNew);
		return NULL;
	}

	ioSockNew->sock		= sockNew;
	ioSockNew->sockAddr	= addrNew;

	//dMemcpy(&(tcps->sockAddr), &addrNew, sizeof(addrNew));
		
	return ioSockNew;
}


void * RPGAPI NetTCPModule::Connect(char *svTarget)
{
	//新建socket
	// Create socket
	NetSocket sock;
	sock = Net::openSocket();// socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
	if(sock == InvalidSocket)
		return NULL;

	//Net::setOption(sock, Net::N_TCP_NODELAY, TRUE, Net::N_IPPROTO_TCP);
	//Net::setOption(sock, Net::N_SO_DONTLINGER, TRUE);

	NetAddress  netAddr;
	Net::stringToAddress(svTarget, &netAddr);
	

	//连接远程主机
	// Connect to remote host
	if( Net::connect(sock, &netAddr) != Net::NoError) 
	{
		Net::closeSocket(sock);
		return NULL;
	}
	
	//返回TCPSocket信息
	// Allocate internal state structure
	//新建IO Socket对象
	NETTCPSOCKET *ioSockNew = gs_netTCPSocketPool.alloc();//(NETTCPSOCKET*)dMalloc(sizeof(NETTCPSOCKET));
	if(ioSockNew == NULL) 
	{
		Net::closeSocket(sock);
		return NULL;
	}

	ioSockNew->sock		= sock;
	ioSockNew->sockAddr	= netAddr;
	//dMemcpy(&(tcps->sockAddr), &addrNew, sizeof(addrNew));
		
	return ioSockNew;
	
}



int RPGAPI NetTCPModule::Close(void *ios)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)ios;
	
	Net::closeSocket(tcps->sock);

	gs_netTCPSocketPool.free(tcps);
	return 0;
}


char * RPGAPI NetTCPModule::Query()
{
	return "NetTCP TGE Net TCP Module v1.0";
}



BOOL RPGAPI NetTCPModule::CheckRecv(void *data)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET *)data;

	return Net::checkRecv(tcps->sock,0,0) == Net::NoError;

	// Check socket for readability 
	//TIMEVAL	tv = {0,0};
	//fd_set	rfds;
	//int		nRet;

	//FD_ZERO(&rfds);
	//FD_SET(tcps->sock,&rfds);
	//nRet = select(1,&rfds,NULL,NULL,&tv);//WSAENOTINITIALISED

	//return (nRet > 0);
}


int RPGAPI NetTCPModule::Recv(void *data, BYTE *pInData, int *pnInDataLen)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET *)data;
	//获取数据长度
	// Get length of rest of data
	S32	nRet;

	S32	nLenRecv;

	//确保有数据存在
	// Make sure we have the rest of the packet
	U32 nPacketSize;
	nRet = Net::ioCtrl(tcps->sock, Net::N_FIONREAD, &nPacketSize);
	if(nRet != Net::NoError) 
		return Net::IOFailed;

	if(nPacketSize > MaxPacketDataSize) 
		return Net::IOInvalid;



	U8	*pBuffer = pInData;
	//接受数据包
	// Receive data
	U8 *pBufRecv	= (U8*) pBuffer;
	int nLeftNum	= nPacketSize;
	S32	nTrys		= 0;
	do 
	{
		nRet = Net::recv(tcps->sock, pBufRecv, nLeftNum, &nLenRecv);

		//如果packetSize为0，则需要重新获取数据包大小
		if(nPacketSize == 0)
		{
			nRet = Net::ioCtrl(tcps->sock, Net::N_FIONREAD, &nPacketSize);
			if(nRet == Net::NoError && nPacketSize != 0)
			{	
				if(nPacketSize > MaxPacketDataSize) 
					return Net::IOInvalid;
				nLeftNum = nPacketSize;
				continue;
			}
			else
				nLenRecv = 0;
		}

		//如果接收数量为0，视为断线
		if(nLenRecv == 0) 
		{	
			*pnInDataLen	= 0;
			return Net::IODisconnected;
		}

		if(nRet != Net::NoError) 
		{
			*pnInDataLen	= 0;
			return Net::IOFailed;
		}

		nLeftNum	-= nLenRecv;
		pBufRecv	+= nLenRecv;
		if(nLeftNum > 0)
			Platform::sleep(RECV_IDLE);

	} while(nLeftNum > 0);

	// Pass data back to application
	//*pInData		= pBuffer;
	*pnInDataLen= nPacketSize;
	return Net::IOOK;
}


int RPGAPI NetTCPModule::RecvEx(void *data, BYTE **pInData, int *pnInDataLen)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET *)data;
	//获取数据长度
	// Get length of rest of data
	int	nRet;
	int	nPacketSize;
	int	nLenRecv;
	nRet = Net::recv(tcps->sock, (U8*)&nPacketSize, sizeof(int), &nLenRecv, Net::N_MSG_PEEK);
	if(nRet != Net::OK) 
		return Net::IOFailed;


	if(nLenRecv < sizeof(int)) 
		return Net::IOInvalid;

	//确保有数据存在
	// Make sure we have the rest of the packet
	U32 len;
	nRet = Net::ioCtrl(tcps->sock, Net::N_FIONREAD, &len);
	if(nRet != Net::NoError) 
		return Net::IOFailed;

	if(len < (sizeof(int) + nPacketSize)) 
		return Net::IOInvalid;

	//请除头信息
	// Clear off the header
	nRet = Net::recv(tcps->sock, (U8*)&nPacketSize, sizeof(int), &nLenRecv);
	if(nLenRecv < sizeof(int)) 
		return Net::IOFailed;
	
	//申请数据缓冲
	// Allocate buffer for data
	//AssertWarn(0,"需要进行内存池管理");
	BYTE	*pBuffer = (BYTE*) dMalloc(nPacketSize);
	if(pBuffer == NULL) 
	{
		*pInData		= NULL;
		*pnInDataLen= 0;
		return Net::IOFailed;
	}


	//接受数据包
	// Receive data
	U8 *pBufRecv	= (U8*) pBuffer;
	int nLeftNum	= nPacketSize;
	do 
	{
		nRet = Net::recv(tcps->sock, pBufRecv, nLeftNum, &nLenRecv);

		if(nRet != Net::NoError || nLenRecv == 0) 
		{
			dFree(pBuffer);
			*pInData		= NULL;
			*pnInDataLen= 0;
			//如果接收数量为0，视为断线
			if(nLenRecv == 0)
				return Net::IODisconnected;
			return Net::IOFailed;
		}
		nLeftNum	-= nLenRecv;
		pBufRecv	+= nLenRecv;
		if(nLeftNum > 0)
			Platform::sleep(RECV_IDLE);

	} while(nLeftNum > 0);

	// Pass data back to application
	*pInData		= pBuffer;
	*pnInDataLen= nPacketSize;
	return Net::IOOK;
}


BOOL RPGAPI NetTCPModule::CheckSend(void *data)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)data;
	//发送数据包
	return Net::checkSend(tcps->sock,SEND_TIMEOUT,0) == Net::NoError;

	// Send packet
	//TIMEVAL	tm;
	//fd_set	wfds;
	//int		ret;

	//tm.tv_sec	= SEND_TIMEOUT;
	//tm.tv_usec	= 0;

	//FD_ZERO(&wfds);
	//FD_SET(tcps->sock,&wfds);

	//ret = select(0,NULL,&wfds,NULL,&tm);
	//return (ret > 0);
}


int RPGAPI NetTCPModule::Send(void *data, BYTE *pData, int nDataLen)
{
	AssertWarn(nDataLen <= MaxPacketDataSize, "需要小于最大包大小...");
	S32		nSendLen;
	S32		nRet;
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)data;

	//生成单一数据包
	// Make single packet
	void *pBufPacket = pData;//dMalloc( sizeof(int) + nDataLen);
	if(pBufPacket == NULL) 
		return Net::IOFailed;

	//填充数据包长度、数据包内容
	// Send packet length
	//dMemcpy(pBufPacket, &nDataLen, sizeof(int));
	//dMemcpy((BYTE*)pBufPacket + sizeof(int), pData, nDataLen);
	
	{
		U8	*pBufSend		= (U8*)pBufPacket;
		int	nLeftNum		= nDataLen;//+sizeof(int);
		do 
		{
			nRet = Net::send(tcps->sock, pBufSend, nLeftNum, &nSendLen);
			if(nRet != Net::NoError)
				break;

			nLeftNum	-= nSendLen;
			pBufSend	+= nSendLen;
			if(nLeftNum > 0)
				Platform::sleep(SEND_IDLE);

		} while(nLeftNum > 0);

		//dFree(pBufPacket);
		if(nRet != Net::NoError)
			return Net::IOFailed;
		return Net::IOOK;
	}

	//dFree(pBufPacket);
	return nSendLen;
}


int RPGAPI NetTCPModule::SendEx(void *data, BYTE *pData, int nDataLen)
{
	S32	nSendLen;
	S32	nRet;
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)data;

	//生成单一数据包
	// Make single packet
	void *pBufPacket = dMalloc( sizeof(int) + nDataLen);
	if(pBufPacket == NULL) 
		return Net::IOFailed;

	//填充数据包长度、数据包内容
	// Send packet length
	dMemcpy(pBufPacket, &nDataLen, sizeof(int));
	dMemcpy((BYTE*)pBufPacket + sizeof(int), pData, nDataLen);
	
	{
		U8	*pBufSend		= (U8*)pBufPacket;
		int	nLeftNum		= nDataLen+sizeof(int);
		do 
		{
			nRet = Net::send(tcps->sock, pBufSend, nLeftNum, &nSendLen);
			 if(nRet != Net::NoError)
				break;

			nLeftNum	-= nSendLen;
			pBufSend	+= nSendLen;
			if(nLeftNum > 0)
				Platform::sleep(SEND_IDLE);

		} while(nLeftNum > 0);

		dFree(pBufPacket);
	 if(nRet != Net::NoError)
			return Net::IOFailed;
		return Net::IOOK;
	}

	dFree(pBufPacket);
	return nSendLen;
}


void RPGAPI NetTCPModule::Free(void *data, BYTE *pBuffer)
{
	if(pBuffer==NULL) 
		return;
	dFree(pBuffer);
}


void* RPGAPI NetTCPModule::GetNetAddress(void *data)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)data;
	return &tcps->sockAddr;
}

int RPGAPI NetTCPModule::GetConnectAddr(void *data, char *svAddr, int nMaxLen)
{
	NETTCPSOCKET *tcps = (NETTCPSOCKET*)data;

	if(nMaxLen < 0)
		return Net::IOFailed;
	if(nMaxLen > ADDR_MAX) 
		nMaxLen = ADDR_MAX;

	char szAddr[256];
	Net::addressToString(&tcps->sockAddr,szAddr);
	dStrncpy(svAddr, szAddr, nMaxLen);
	return Net::IOOK;
}


};//namespace CS