socketex.cpp

Socket异步通信,线程,双端队列 网络实时通信。

// SocketEx.cpp : implementation file
//

#include "stdafx.h"
#include "Socket.h"
#include "SocketEx.h"

#ifdef _DEBUG
#define new DEBUG_NEW
#undef THIS_FILE
static char THIS_FILE[] = __FILE__;
#endif

/////////////////////////////////////////////////////////////////////////////
// CSocketEx

CSocketEx::CSocketEx(CWnd* pOwner,UINT nPort,BOOL bRecv,CString strIp) : CSocket()
{
	//防止中间频繁的分配和释放内存;
	const long nBufLen0 = 4096;	//定义缓冲的最大长度；
	m_nBufLen0 = nBufLen0;
	m_cBufferR = new char[m_nBufLen0+1];  //便于版本升级//字符串需要空字符结束;
	m_nLengthR = 0;
	memset(m_cBufferR,0,m_nBufLen0+1);
	
	m_pThread     = 0;
	m_hEventAr[0] = 0;
	m_hEventAr[1] = 0;
	m_hEventAr[2] = 0;

	Init2(pOwner,nPort,bRecv,strIp);
}

CSocketEx::~CSocketEx()
{
	StopAction();	//可能需要结束线程;
	//后删除
	delete [] m_cBufferR;
}

void CSocketEx::Init2(CWnd* pOwner,UINT nPort,BOOL bRecv,CString strIp)
{
	m_pOwner=pOwner;
	m_bRecv = bRecv;
	m_nPort = nPort;
	m_strIp = strIp;
	if(strIp=="")
	{
		CString strCp;
		m_strIp = GetIpAddress(strCp);	//缺省时，自动获取Ip；	
	}
}


// Do not edit the following lines, which are needed by ClassWizard.
#if 0
BEGIN_MESSAGE_MAP(CSocketEx, CSocket)
	//{{AFX_MSG_MAP(CSocketEx)
	//}}AFX_MSG_MAP
END_MESSAGE_MAP()
#endif	// 0

/////////////////////////////////////////////////////////////////////////////
// CSocketEx member functions

void CSocketEx::OnAccept(int nErrorCode) 
{
	// TODO: Add your specialized code here and/or call the base class
	//侦听到连接请求，调用Accept函数
	CSocket* pASocket = new CSocket();
	if(Accept(*pASocket))
	{
		pASocket->AsyncSelect(FD_READ);
		
		m_FifoX.rpush( pASocket ); //这些new分配的对象由队列来管理;
	}
	else
	{
		delete pASocket;
	}

	CSocket::OnAccept(nErrorCode);
}

void CSocketEx::OnClose(int nErrorCode) 
{
	// TODO: Add your specialized code here and/or call the base class
	if(m_hSocket!=INVALID_SOCKET)
		Close();
	AsyncSelect(FD_CLOSE);
	
	CSocket::OnClose(nErrorCode);
}

void CSocketEx::OnReceive(int nErrorCode) 
{
	// TODO: Add your specialized code here and/or call the base class
	try
	{
		CString strIp;
		unsigned int nPort;
		m_nLengthR = ReceiveFrom(m_cBufferR,m_nBufLen0,strIp,nPort,0);
		//回发一个接收成功的消息：
		if(strIp!="")  //如果strIp为空串,则接收到其为端口主动发出的消息;
		{ 
			m_cBufferR[m_nLengthR] = '\0';
			CFifoY<CByteArray,CString>* o = new CFifoY<CByteArray,CString>;
			CByteArray* btA = o->GetpValX();
			btA->SetSize(0);
			btA->SetSize(m_nLengthR);
			BYTE* bt = btA->GetData();
			memcpy(bt,m_cBufferR,m_nLengthR);
			//完成消息内容存放;
			o->SetValY(strIp);	//保存Ip地址;
			o->SetvSize(nPort);	//保存端口号;//下一步需要回发消息,但不能在此处处理;
			m_Fifo2.rpush( o );
			
			//if(m_hEventAr[1]!=NULL)	ResetEvent(m_hEventAr[1]);	//不能放此位置；

			SetEvent(m_hEventAr[1]);
			//警告：下面两行代码放到线程中处理,否则逻辑上就形成收发端无限死循环！
			//char   ss[7];  ss[0]='S';ss[1]='U';ss[2]='C';ss[3]='C';ss[4]='E';ss[5]='S';ss[6]='S';
			//SendTo(ss,7,nPort,strIp);
		}
	}
	catch(...)
	{
	}
	AsyncSelect(FD_WRITE);
	
	CSocket::OnReceive(nErrorCode);
}

void CSocketEx::OnSend(int nErrorCode) 
{
	// TODO: Add your specialized code here and/or call the base class
	try
	{   
		//此处添加执行除发送以外的处理代码;//代码完整性;
		SetEvent(m_hEventAr[2]);
	}
	catch(...)
	{
	}
	AsyncSelect(FD_READ);
	
	CSocket::OnSend(nErrorCode);
}

void CSocketEx::SetEvent1()
{
	SetEvent(m_hEventAr[1]);
}

void CSocketEx::SetEvent2()
{
	SetEvent(m_hEventAr[2]);
}

void CSocketEx::InitEvent()
{
	const int N = 3;
	for(int i=0;i<N;i++)
	{
		if(m_hEventAr[i]!=NULL)	ResetEvent(m_hEventAr[i]);
		m_hEventAr[i] = CreateEvent(NULL,TRUE,FALSE,NULL);
	}
}

UINT CSocketEx::ThreadSocketEx(LPVOID pParam)
{
	CSocketEx* pT = (CSocketEx*)pParam;
	DWORD nEvent = 0;
	for(;;)
	{
		nEvent = WaitForMultipleObjects(3,pT->m_hEventAr,FALSE,INFINITE);	//INFINITE//要求无限等待；

		switch (nEvent)
		{
			case WAIT_OBJECT_0  :
			{
				ResetEvent(pT->m_hEventAr[0]);
				//接到结束线程通知:
				DWORD dwExitCode;
				GetExitCodeThread(pT->m_pThread->m_hThread,&dwExitCode);
				pT->m_pThread = 0;
				TRACE("Exit SocketEx-Thread.\n");
				AfxEndThread( dwExitCode);
				break;	//本行不会执行的；
			}
			case WAIT_OBJECT_0+1:
			{
				ResetEvent(pT->m_hEventAr[1]);
				//接到接收事件通知:
				pT->ActionRecv();
				break;
			}
			case WAIT_OBJECT_0+2:
			{
				ResetEvent(pT->m_hEventAr[2]);
				//接到发送事件通知:
				pT->ActionSend();
				break;
			}
		}
	}
	
	pT->m_pThread = 0;	//线程自动退出，需作出标记；
	return 0;
}

BOOL CSocketEx::ActionRecv()
{
	if(m_Fifo2.bHasNode())	//因为每个消息都会通知一次；
	{
        try	{
///////////////////////////////////////////////////////////////////////////////
		CFifoY<CByteArray,CString>* odd = m_Fifo2.lpop();
        CByteArray* btA = odd->GetpValX();
		CString strIp = odd->GetValY();
		UINT nPort = odd->GetvSize();

		//在此处把接收的消息发送到窗口!!!
		if(m_pOwner&&m_pOwner->m_hWnd)
		{	//特别注意：节点odd在消息被发送的窗口要使用时,『使用其数据域』
			//将odd添加到节点删除队列，以便在线程终止的时候，由程序控制将这些节点删除；减少用户的中间操作，所以在此处进行必要的容错处理；
			m_FifoT.rpush( odd );	//
			::SendMessage(m_pOwner->m_hWnd, WM_SOCKETEX, (WPARAM) odd, (LPARAM) 0);	//接收消息通知！注意：此处的odd不要带地址符号；
		}
		else
		{	//如果不发送消息到指定窗口，则在此处进行处理，直接将该弹出节点删除就是；
			delete odd;odd=0;
		}
        
		if( !(nPort==m_nPort&&strIp==m_strIp) )	//如果不是端口自己给自己发消息才执行回发操作！！！
		{
			int nSize = btA->GetSize();
			BYTE* bt = btA->GetData();
			if(nSize>0 && *(bt+nSize-1)!=0)	//说明收到的消息不是对方发给自己的回复消息!
			{
				//此处为回发消息代码://测试如：char   ss[7];  ss[0]='S';ss[1]='U';ss[2]='C';ss[3]='C';ss[4]='E';ss[5]='S';ss[6]='S';
				char* s = new char[nSize +1 ];	//给回复消息长度增加1！
				for(int r=0;r<nSize;r++)
					*(s+r) = (char)(*(bt+r));	//原数据本为char*类型；
				*(s+nSize) = 0;
				SendTo(s,nSize+1,nPort,strIp);	//消息回发末尾多补一0！《供双方判断》
				delete [] s;	//立即删除中间的new对象；
				////SetEvent(m_hEventAr[2]);		//此处设置发送消息事件是错误的//不要该行代码，因为回复是自动的；而且SendTo-也可能导致该事件；
			}
		}
///////////////////////////////////////////////////////////////////////////////
		}	catch(...)
		{
			return FALSE;
		}
	}
	
	return TRUE;
}

BOOL CSocketEx::ActionSend()
{
	//与应用程序有关的回发消息处理,或为发送端口主动发送消息处理;通常此块代码是不会执行的；
	if(m_pOwner&&m_pOwner->m_hWnd)
	{
		::SendMessage(m_pOwner->m_hWnd, WM_SOCKETEX, (WPARAM) 0, (LPARAM) 1);	//发送消息通知！
	}

	return TRUE;
}

UINT CSocketEx::StarAction()
{
	try	{
		BOOL bOpenSucc = false;
		if(m_bRecv)
			bOpenSucc = Create(m_nPort,SOCK_DGRAM,m_strIp);
		else
			bOpenSucc = Create(m_nPort,SOCK_DGRAM,NULL   );
		if(!bOpenSucc )	return 0;

		Bind(m_nPort,m_strIp);
	}
	catch(...)	{	return 0;	}	//本次端口打开失败;//警告无法捕获异常!
			
	if(!m_bRecv){	return 1;	}	//发送端口打开成功;//不是接收端口返回;

	//如果是接收端口则执行启动线程代码;
	if(!m_pThread)
	{
///////////////////////////////////////////////////////////////////////////////
		//执行事件重新初始化:[中间被改变],然后才可以进入线程启动状态;
		InitEvent();
		//线程启动;
		if( !(m_pThread = AfxBeginThread(ThreadSocketEx, this)) )	//创建线程失败!
			return 1;
		TRACE("SocketEx-Thread started\n");
///////////////////////////////////////////////////////////////////////////////
		//线程打开后并无事件通知;
///////////////////////////////////////////////////////////////////////////////
		return 3;	//线程启动成功!
	}
	return 1;		//只是打开成功!
}

void CSocketEx::StopAction()
{
//这两行与线程无关！
	m_FifoX.pClear();  //new-节点值域;//需要主动调用,清空指针对象;
	m_FifoX.Clear();
//上两行与线程无关！

///////////////////////////////////////////////////////////////////////////////
	//if(!m_pThread)	return;	//等待的线程！堆对象此前释放有此行代码,发送端口无法Close
	//结束线程通知:优先级最高!
	int N = 0;
	while(m_pThread)
	{	N++;
		SetEvent(m_hEventAr[0]);	//触发事件，同时此循环代码在执行，故需延时！
		Sleep( 10 );	//等待足够的时间，以让线程代码得以执行，以至超过计数器！
		if(N==100 )		//成功打开非法串口导致事件触发无效，用手工强行终止线程！
		{
			TerminateThread(m_pThread->m_hThread,259);
			m_pThread = 0;
			TRACE("Exit SocketEx-Thread.\n");
			break;
		}
	}
	if(m_hSocket!=INVALID_SOCKET)
		Close();

	m_FifoT.pClear();  //弹出节点地址;//需要主动调用,清空指针对象;
	m_FifoT.Clear();
}

CString CSocketEx::GetIpAddress(CString& sCpName)
{
	//获取本机IP地址
	static char FAR name[128];
	gethostname(name, sizeof(name));	sCpName = name;
	struct hostent FAR * pHostent = gethostbyname(name);
	if(!pHostent)	return "";	//"127.0.0.1";//说明无法进行Socket通信!

	LPCSTR ip = inet_ntoa( *(struct in_addr *)*pHostent->h_addr_list );
	return (CString)ip; 
}