baseserver.cpp

A3服务端AccountServer源代码

#include "stdinc.h"

#include "BaseServer.h"
#include "LoopSleepMgr.h"

CBaseServer::CBaseServer()
{
	m_bConnect = FALSE;
	m_nRecvStartPos = m_nRecvEndPos = 0;

	m_nSendStartPos = m_nSendEndPos = 0;

	m_wsaEvent = WSACreateEvent();
	InitializeCriticalSection(&m_csSend);
}

CBaseServer::~CBaseServer()
{
	WSACloseEvent(m_wsaEvent);
	DeleteCriticalSection(&m_csSend);
}


BOOL	CBaseServer::Connect(char * pszAddr, int nPort)
{
	if ((m_sSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET)
	{
		return FALSE;
	}

	SOCKADDR_IN		sinSockArr; 
	sinSockArr.sin_family = AF_INET;
	sinSockArr.sin_addr.s_addr = inet_addr(pszAddr); 
	sinSockArr.sin_port = htons(nPort);

	// sync socket 栏肺 connect 矫档 
	int n = connect(m_sSocket, (SOCKADDR *)&sinSockArr, sizeof(sinSockArr));
	if (n < 0 || n == SOCKET_ERROR)
	{
		closesocket(m_sSocket);
		return FALSE;
	}	
	
	m_bConnect = TRUE;
	return TRUE;
}

void CBaseServer::SetNetEvent()
{
	WSAEventSelect(m_sSocket, m_wsaEvent, FD_READ | FD_CLOSE);  
}

void	CBaseServer::SvrProc()
{
	if (!m_bConnect) return;

	if (WSAWaitForMultipleEvents(1, &m_wsaEvent, FALSE, 0, TRUE) == WSA_WAIT_EVENT_0)
	{
		WSAEnumNetworkEvents(m_sSocket, m_wsaEvent, &m_netEvent);

		if (m_netEvent.lNetworkEvents & FD_READ)
		{
			if (RecvMsg())
			{
				ParseMsg();
			}
			else
			{
				printf("////////////////////////////\n  FD_READ ERROR : Close\n////////////////////////////\n");
				m_bConnect = FALSE;
				OnClose();			
			}
		}

		if (m_netEvent.lNetworkEvents & FD_CLOSE)
		{
			printf("////////////////////////////\n  FD_CLOSE ERROR : Close\n////////////////////////////\n");
			m_bConnect = FALSE;
			OnClose();			
		}
	}
}

BOOL	CBaseServer::RecvMsg()
{
	int		n, nErr;
	char 	MsgBuf[MAX_RECV_BUF];
	
	if ((n = recv(m_sSocket, MsgBuf, MAX_RECV_BUF, 0)) > 0) 
	{
		return (PushMsgQ(MsgBuf, n));
	}
	else if (n == SOCKET_ERROR)
	{
		if ((nErr = WSAGetLastError()) == WSAEWOULDBLOCK)
		{
			printf ("Recv WSAEWOULDBLOCK\n");
			return TRUE;
		}
	}

	return FALSE;
}


BOOL	CBaseServer::SendMsg(void * pMsg, int nLen)
{
/*	int		Err;
	Err = send(m_sSocket, (const char FAR * )pMsg, nLen, 0);

	if (Err != 0 || (WSAGetLastError() == WSAEWOULDBLOCK))
	{
		return TRUE;
	}

	return FALSE;
*/
	if (m_bConnect == FALSE) return FALSE;
	return PushSendMsgQ(pMsg, nLen);
}


//----------------------------------------------------------------------------------
// recv 茄 msg Q 包府 
//----------------------------------------------------------------------------------

int		CBaseServer::GetSizeInBuf()
{
	return (m_nRecvEndPos - m_nRecvStartPos);
}

BOOL	CBaseServer:: PushMsgQ(void * pMsg, int nLen)
{
	if ((nLen + m_nRecvEndPos) > MAX_MSG_Q_BUF)
	{
		return FALSE;
	}

	memcpy(&(m_RecvMsgQBuf[m_nRecvEndPos]), pMsg, nLen);
	m_nRecvEndPos += nLen;

	return TRUE;
}

BOOL	CBaseServer::PopMsgQ(void * pMsg, int nLen)
{
	if (GetSizeInBuf() < nLen) return FALSE; 

	memcpy(pMsg, &(m_RecvMsgQBuf[m_nRecvStartPos]), nLen);
	m_nRecvStartPos += nLen;
	if (m_nRecvStartPos == m_nRecvEndPos) m_nRecvStartPos = m_nRecvEndPos = 0;

	return TRUE;
}

void *	CBaseServer::PopMsgQ(int nLen)
{
	int nSize; 
	if ((nSize = GetSizeInBuf()) < nLen)
	{
		if (m_nRecvEndPos >= PROTECT_MSG_Q_BUF)
		{
			memcpy(m_RecvMsgQBuf, &(m_RecvMsgQBuf[m_nRecvStartPos]), nSize);
			m_nRecvStartPos = 0;
			m_nRecvEndPos = nSize;
		}
		return NULL; 
	}

	void	* pMsg = &(m_RecvMsgQBuf[m_nRecvStartPos]);
	m_nRecvStartPos += nLen;
	if (m_nRecvStartPos == m_nRecvEndPos) m_nRecvStartPos = m_nRecvEndPos = 0;

	return pMsg;
}

//----------------------------------------------------------------------------------
// send thread Q 包府 
//----------------------------------------------------------------------------------

int		CBaseServer::GetSizeInSendBuf()
{
	return (m_nSendEndPos - m_nSendStartPos);
}

BOOL	CBaseServer::PushSendMsgQ(void * pMsg, int nLen)
{
	if ((nLen + m_nSendEndPos) > MAX_MSG_Q_BUF) return FALSE;

	Lock();
	memcpy(&(m_SendMsgQBuf[m_nSendEndPos]), pMsg, nLen);
	m_nSendEndPos += nLen;
	Unlock();

	return TRUE;
}

BOOL	CBaseServer::InitSendThread()
{
	m_nSendStartPos = m_nSendEndPos = 0;	

	if (BeginThread() == FALSE) return FALSE;

	return TRUE;
}


void	CBaseServer::LoopProc()
{
	CLoopSleepMgr	SendLoopSleepMgr;
	SendLoopSleepMgr.SetValue(2);		// Loop = 2 * 64 雀/檬 

	m_bRunThread = TRUE;
	while (m_bRunThread)
	{
		SendProc();

		SendLoopSleepMgr.LoopSleep();
	}
}

void	CBaseServer::SendProc()
{
	int n, nSendSize; 

	if (GetSizeInSendBuf())
	{
		Lock();
		
		nSendSize = MIN(1024*64, GetSizeInSendBuf());
		n = send(m_sSocket, &(m_SendMsgQBuf[m_nSendStartPos]), nSendSize, 0);
		if ((n > 0) && (n != SOCKET_ERROR))
		{
			m_nSendStartPos += n;
			if (m_nSendStartPos == m_nSendEndPos) m_nSendStartPos = m_nSendEndPos = 0;
			else if (m_nSendEndPos >= PROTECT_MSG_Q_BUF)
			{
				int nSize = GetSizeInSendBuf();
				memcpy (m_SendMsgQBuf, &(m_SendMsgQBuf[m_nSendStartPos]), nSize);
				m_nSendStartPos = 0;
				m_nSendEndPos = nSize;
			}
		}
		else
		{
			// 俊矾!
			printf ("SEND ERROR : %d  (Conn=%d, buff=%d)\n", WSAGetLastError(), m_bConnect, GetSizeInSendBuf());
		}

		Unlock();
	}
}

void	CBaseServer::CloseSendThread()
{
	CLoopThread::CloseThread();
}


int		CBaseServer::ResizeSNDBUF(const int nSize)
{
	return setsockopt( m_sSocket, SOL_SOCKET, SO_SNDBUF, (char *)&nSize, sizeof(nSize) );
}

int		CBaseServer::ResizeRCVBUF(const int nSize)
{
	return setsockopt( m_sSocket, SOL_SOCKET, SO_RCVBUF, (char *)&nSize, sizeof(nSize) );
}