lyfz_repos/经典版源码/kernelcode/lyfzWeiXin/NetWork/ClientProcess.cpp

#include "StdAfx.h"
#include "ClientProcess.h"
#include <comdef.h>
#include <atlbase.h>

const int AF_IPV4   = 0;
const int AF_IPV6   = 1;
const int SOCK_TCP  = SOCK_STREAM-1;
const int SOCK_UDP  = SOCK_DGRAM-1;


//CClientProcess* CClientProcess::m_pTcpClient[TCPCLIENTNUM] = {NULL};
void GetFileName(IN const TCHAR *pFullName,OUT TCHAR *pFileName)
{
	TCHAR szExt[_MAX_EXT];
	_tsplitpath(pFullName, NULL, NULL, pFileName, szExt);
	_tcscat(pFileName, szExt);
}

CClientProcess::CClientProcess():m_nMode(AF_IPV4),m_nSockType(SOCK_TCP) 
{
	m_bSocket = FALSE;
	m_bStopbeat = FALSE;
	m_hRunObject = NULL;
	m_bRecevie = TRUE;
	m_hReConnectSrvThreadHandle = NULL;
	m_dwSumRecive = 0;
	m_dwCurRecive = 0;

	m_SocketClient.SetInterface(this);
}

CClientProcess::~CClientProcess()
{
	//m_SocketClient.Terminate();
	net_CloseSocket();
}

BOOL CClientProcess::SolveError()
{
	DWORD dwWSAError = WSAGetLastError();

	if ( dwWSAError != 0 )
	{
		//LOG4C_NO_FILENUM((LOG_NOTICE,"dwWSAError = %d~",dwWSAError));
	}
	
	switch(dwWSAError)
	{
	case WSAENOTSOCK:
	case WSAENETDOWN:
	case WSAENETUNREACH:
	case WSAENETRESET:
	case WSAECONNABORTED:
	case WSAECONNRESET:
	case WSAESHUTDOWN:
	case WSAEHOSTDOWN:
	case WSAEHOSTUNREACH:
		TRACE("-----------------WSAError = %d\n",dwWSAError);
		//LOG4C_NO_FILENUM((LOG_NOTICE,"dwWSAError = %d~",dwWSAError));
		m_bSocket = FALSE;
		break;
	default:
		break;
	}

	return TRUE;
}

void CClientProcess::GetAddress(const SockAddrIn& addrIn, CString& rString) const
{
	TCHAR szIPAddr[MAX_PATH] = { 0 };
	CSocketHandle::FormatIP(szIPAddr, MAX_PATH, addrIn);
	rString.Format(_T("%s : %d"), szIPAddr, static_cast<int>(static_cast<UINT>(ntohs(addrIn.GetPort()))) );
}

void CClientProcess::AppendText(LPCTSTR lpszFormat, ...)
{
	//	if ( !::IsWindow(m_ctlMsgList.GetSafeHwnd()) ) return;
	//	TCHAR szBuffer[512];
	//	HWND hWnd = m_ctlMsgList.GetSafeHwnd();
	//	DWORD dwResult = 0;
	//	if (SendMessageTimeout(hWnd, WM_GETTEXTLENGTH, 0, 0, SMTO_NORMAL, 500L, &dwResult) != 0)
	//	{
	//		int nLen = (int) dwResult;
	//		if (SendMessageTimeout(hWnd, EM_SETSEL, nLen, nLen, SMTO_NORMAL, 500L, &dwResult) != 0)
	//		{
	//			size_t cb = 0;
	//			va_list args;
	//			va_start(args, lpszFormat);
	//			::StringCchVPrintfEx(szBuffer, 512, NULL, &cb, 0, lpszFormat, args);
	//			va_end(args);
	//			SendMessageTimeout(hWnd, EM_REPLACESEL, FALSE, reinterpret_cast<LPARAM>(szBuffer), SMTO_NORMAL, 500L, &dwResult);
	//		}
	//	}
}

bool CClientProcess::GetDestination(SockAddrIn& addrIn) const
{
	CString strPort;
	int nFamily = (m_nMode == AF_IPV4) ? AF_INET : AF_INET6;
	return addrIn.CreateFrom(NULL, strPort, nFamily);
}

bool CClientProcess::SetupMCAST()
{
	const TCHAR szIPv4MCAST[] = TEXT("239.121.1.2");
	const TCHAR szIPv6MCAST[] = TEXT("FF02:0:0:0:0:0:0:1"); // All Nodes local address
	bool result = false;
	if ( m_nSockType == SOCK_UDP )
	{
		if ( m_nMode == AF_IPV4 ) {
			result = m_SocketClient->AddMembership(szIPv4MCAST, NULL);
		} else {
			result = m_SocketClient->AddMembership(szIPv6MCAST, NULL);
			HRESULT hr = HRESULT_FROM_WIN32(GetLastError());
			hr = hr;
		}
	}
	return result;
}

///////////////////////////////////////////////////////////////////////////////

void CClientProcess::OnThreadBegin(CSocketHandle* pSH)
{
	ASSERT( pSH == m_SocketClient );
	(pSH);
	CString strAddr;
	SockAddrIn sockAddr;
	m_SocketClient->GetSockName(sockAddr);
	GetAddress( sockAddr, strAddr );
}

void CClientProcess::OnThreadExit(CSocketHandle* pSH)
{
	ASSERT( pSH == m_SocketClient );
	(pSH);
}

void CClientProcess::OnConnectionDropped(CSocketHandle* pSH)
{
	ASSERT( pSH == m_SocketClient );
	m_bSocket = FALSE;
	AppendText( _T("Connection lost with client.\r\n") );
}

void CClientProcess::OnConnectionError(CSocketHandle* pSH, DWORD dwError)
{
	ASSERT( pSH == m_SocketClient );
	(pSH);
	_com_error err(dwError);
	AppendText( _T("Communication Error:\r\n%s\r\n"), err.ErrorMessage() );
}

/************************************************************************/
/*  函数：OnDataReceived[3/21/2016 IT];
/*  描述：;
/*  参数：;
/*  	[IN] pSH：		客户端实例对象;
/*  	[IN] pbData：	客户端本次接收到的数据;
/*  	[IN] dwCount：	客户端本次接收到的数据长度;
/*		[IN] addr：		服务端地址;
/*  返回：void;
/*  注意：;
/*  示例：;
/*
/*  修改：;
/*  日期：;
/*  内容：;
/************************************************************************/
void CClientProcess::OnDataReceived(CSocketHandle* pSockHandle, const BYTE* pReceivebuf, DWORD dwReceiveSize, const SockAddrIn& addr)
{
	ASSERT( pSockHandle == m_SocketClient );
	if( !m_SocketClient->IsOpen() ) return;
	if (NULL == pReceivebuf) return;

	DWORD dwIndexOfProcessed = 0;		// 当前接收的包被处理的长度;
	DWORD dwOnceProcessedLen = 0;		// 循环一次处理了多少当前包;
	TheProhead *ptphead;

	while( dwIndexOfProcessed < dwReceiveSize)
	{
		ptphead = NULL;
		dwOnceProcessedLen = 0;

		if ( m_dwPendingSize == 0)	//----------------------------------------------// 1.第一次接收或完整组包后接收剩余包;
		{
			if( dwReceiveSize - dwIndexOfProcessed < sizeof(TheProhead) )					// 1.1.第一次接收或剩余数据不足一个包头;
			{
				// 如果第一次接包就没有收够包头;
				m_dwPendingSize = dwReceiveSize - dwIndexOfProcessed;
				//memcpy(pSockHandle->m_szpendingbuf, pstProtocolheader, pSockHandle->m_npendingSize);
				memcpy(m_byPendingbuf, &pReceivebuf[dwIndexOfProcessed], m_dwPendingSize);

				dwOnceProcessedLen = dwReceiveSize - dwIndexOfProcessed;
			}
			else																				// 1.2.第一次接收或剩余数据大于一个包头;
			{
				ptphead = (TheProhead *)&((unsigned char *)pReceivebuf)[dwIndexOfProcessed];
				if( (int)ptphead->nDataLen > dwReceiveSize - dwIndexOfProcessed )		// 1.2.1.第一次接收或剩余数据不足一个完整的协议包;
				{
					//memcpy(pSockHandle->m_szpendingbuf, pstProtocolheader, dwReceiveSize - dwIndexOfProcessed);
					memcpy(m_byPendingbuf, &pReceivebuf[dwIndexOfProcessed], dwReceiveSize - dwIndexOfProcessed);
					m_dwPendingSize = dwReceiveSize - dwIndexOfProcessed;
					dwOnceProcessedLen = dwReceiveSize - dwIndexOfProcessed;	// 该次处理的数据长度;
				}
				else	// ---------------------------------------------------------------------// 1.2.2.第一次接收或剩余数据大于一个完整的协议包;
				{
					//LOG4C((LOG_NOTICE, "正常包"));
					m_dwPendingSize = 0;
					dwOnceProcessedLen = ptphead->nDataLen;	// 该次处理的数据长度;
				}
			}
		}
		else	//------------------------------------------------------------------------------// 2.第n+1次接收数据包;
		{
			if( m_dwPendingSize < sizeof(TheProhead) )	// 2.1.pengingbuf数据小于包头
			{
				DWORD dwRestheader = sizeof(TheProhead) - m_dwPendingSize;
				if( dwRestheader < dwReceiveSize )							// 2.1.1.当前包大小 > 剩下包头长度,可以组成一个完整包头;	
				{
					// -1.先收完一个完整的包头;
					memcpy( &m_byPendingbuf[m_dwPendingSize], pReceivebuf, dwRestheader);

					// -2.处理剩余数据;
					ptphead = (TheProhead *)m_byPendingbuf;
					//memcpy(ptphead, m_byPendingbuf,m_dwPendingSize);
					dwOnceProcessedLen = ptphead->nDataLen - m_dwPendingSize;
					if( dwOnceProcessedLen <= dwReceiveSize )  // 可以组成一个完整的协议包;
					{
						memcpy( &m_byPendingbuf[sizeof(TheProhead)],
							&((char *)pReceivebuf)[dwRestheader],
							ptphead->nDataLen - sizeof(TheProhead));

						m_dwPendingSize = 0;
					}
					else	// 未能组成一个完整的协议包,断续接收等待;
					{
						//int nTemp = dwReceiveSize - dwRestheader; //除去头剩余部分的长度
						//if ( nTemp > 0 )	//刚好是Header的长度，不用拷贝内存，所以这里加了>0的判断
						//{
						//memcpy( &pSockHandle->m_szpendingbuf[sizeof(STProtocolheader)],&((char *)pReceivebuf)[dwRestheader],nTemp );
						//}
						memcpy( &m_byPendingbuf[sizeof(TheProhead)],&((char *)pReceivebuf)[dwRestheader],dwReceiveSize - dwRestheader);
						m_dwPendingSize += dwReceiveSize;
					}
				}
				else //------------------------------------------------------// 2.1.2.当前包大小 <= 剩下包头长度,未能或刚好组成一个完整的包头;
				{
					memcpy( &m_byPendingbuf[m_dwPendingSize], pReceivebuf, dwReceiveSize );
					m_dwPendingSize += dwReceiveSize;
					dwOnceProcessedLen = dwReceiveSize;
				}
			}
			else // --------------------------------------------------------// 2.2.pengingbuf数据大于包头;
			{
				ptphead = (TheProhead *)m_byPendingbuf;
				dwOnceProcessedLen = ptphead->nDataLen - m_dwPendingSize;
				if ( dwOnceProcessedLen <= dwReceiveSize )		// 可以组成一个完整的协议包;
				{
					memcpy( &m_byPendingbuf[m_dwPendingSize], pReceivebuf, dwOnceProcessedLen );
					m_dwPendingSize = 0;
				}
				else // 未能组成一个完整协议包;
				{
					memcpy( &m_byPendingbuf[m_dwPendingSize], pReceivebuf, dwReceiveSize );
					m_dwPendingSize += dwReceiveSize;
				}
			}
		}

		if ( dwOnceProcessedLen == 0 )
		{
			// 没有收够包头，认为是非法包，扔掉
			// LOG4C((LOG_NOTICE, "没有收够包头，认为是非法包，扔掉"));
			break;
		}

		if ( m_dwPendingSize == 0 )
		{
			if ( ptphead->nDataLen > SOCKET_BUFFSIZE )
			{
				// 包长度超过限制,暂时不处理;
				// LOG4C((LOG_NOTICE, "pstProtocolheader->nDataLen超过限制"));
			}

			//if(-1 == OnIntegrityPacket(pSockHandle, ptphead))
		//	{
			//	// LOG4C((LOG_NOTICE, "Error OnIntegrityPacket"));
			//	break;
			//}
		}

		dwIndexOfProcessed += dwOnceProcessedLen;
	}
}

BOOL CClientProcess::Initialize()
{
	TCHAR szIPAddr[MAX_PATH] = { 0 };
	CSocketHandle::GetLocalAddress(szIPAddr, MAX_PATH, AF_INET);
	AppendText(_T("Local Address (IPv4): %s\r\n"), szIPAddr);
	CSocketHandle::GetLocalAddress(szIPAddr, MAX_PATH, AF_INET6);
	AppendText(_T("Local Address (IPv6): %s\r\n"), szIPAddr);

	return TRUE;
}

void CClientProcess::StartReConnectSrvThread()
{
	// Jeff.启用重连服务端线程.-------------------
	m_hRunObject = CreateEvent( NULL, TRUE, FALSE, _T("ClientThreadRun") );
	if ( m_hRunObject == NULL )	
	{
		//LOG4C((LOG_NOTICE,"创建事件失败"));
	}

	m_hReConnectSrvThreadHandle = CreateThread(NULL,0,ReConnectSrvThread,this,0,NULL);
	if ( m_hReConnectSrvThreadHandle == NULL )	
	{
		//LOG4C((LOG_NOTICE,"创建线程失败"));
	}
}

BOOL CClientProcess::ConnectServer(LPCTSTR strAddr, LPCTSTR strPort)
{
	_stprintf_s(m_SvrAddr, MAX_PATH, _T("%s"),strAddr);
	_stprintf_s(m_SvrPort, MAX_PATH, _T("%s"),strPort);
	int nFamily = (m_nMode == AF_IPV4) ? AF_INET : AF_INET6;
	if ( !m_SocketClient.StartClient(NULL, strAddr, strPort, nFamily, (m_nSockType+1) ) )
	{
		m_bSocket = FALSE;		
		return FALSE;
	}

	return TRUE;
}

void CClientProcess::net_CloseSocket()
{
	if(m_hRunObject) 
		SetEvent(m_hRunObject);

	if( m_hReConnectSrvThreadHandle )
	{
		if (WaitForSingleObject(m_hReConnectSrvThreadHandle,INFINITE) != WAIT_FAILED)
		{
			CloseHandle(m_hReConnectSrvThreadHandle);
			m_hReConnectSrvThreadHandle = NULL;
		}
	}

	if(m_hRunObject) 
		CloseHandle( m_hRunObject );
	m_hRunObject = NULL;

	m_SocketClient.Terminate();
}

void CClientProcess::SendMsg(void *pMsg,const int nLen)
{
	if ( m_SocketClient.IsOpen() )
	{
		USES_CONVERSION;
		if (m_nSockType == SOCK_TCP)
		{
			m_SocketClient.Write((const LPBYTE)(pMsg), nLen, NULL, 30000);
		}
		else
		{
			SockAddrIn sockAddr;
			GetDestination(sockAddr);
			m_SocketClient.Write((const LPBYTE)(pMsg), nLen, sockAddr, 30000);
		}
	}
	else
	{
		AfxMessageBox(_T("Socket is not connected"));
	}
}

DWORD WINAPI CClientProcess::ReConnectSrvThread(LPVOID pInstance)
{
	CClientProcess *pClientImpl = (CClientProcess*)pInstance;

	do 
	{
		// 检测连接状态;
		if ( pClientImpl->m_bSocket == FALSE )
		{
			if ( pClientImpl->m_SocketClient->IsOpen() == TRUE )
			{
				pClientImpl->m_SocketClient.Terminate();
			}

			pClientImpl->m_bSocket = pClientImpl->ConnectServer(pClientImpl->m_SvrAddr,pClientImpl->m_SvrPort);
		}

	} while( WaitForSingleObject(pClientImpl->m_hRunObject,200L) == WAIT_TIMEOUT );

	//LOG4C_NO_FILENUM((LOG_NOTICE,"重连服务器线程退出"));

	return 0;
}

INT CClientProcess::Send(IN TheProbody& tpbody, IN const int& nTimeout)
{
	return m_SocketClient->Write((LPBYTE)&tpbody, sizeof(TheProhead) + tpbody.tphead.nDataLen, NULL, nTimeout);
}