hlistener2.h

积下的一点C++编程序库源码

// HListener2.h: interface for the HListener2 class.
//
//////////////////////////////////////////////////////////////////////

#if !defined(AFX_HLISTENER2_H__BF0D5A93_4173_4C8F_B20D_E03405CC4D7E__INCLUDED_)
#define AFX_HLISTENER2_H__BF0D5A93_4173_4C8F_B20D_E03405CC4D7E__INCLUDED_

#if _MSC_VER > 1000
#pragma once
#endif // _MSC_VER > 1000
#include "../uni_include/HTcpSock.h"
#include "../uni_include/UniThread.h"
#include "../uni_include/HMutex.h"
#include "HListenerCall.h"
#include <list>
#include <iostream>
using namespace std ;


class HListener2  
{
	CHTcpSock			m_Sock;
	UNI_BOOL			m_bStop;
	UNI_DWORD			m_RecvThreadCount;
	HLISTENERCALLBACK	m_pUserFunc;
	void *				m_pUserParam;

	UNI_ThreadHandle	m_hListenThread;
	typedef struct HRecvPool
	{
		HRecvPool()
		{
			pThis=UNI_NULL;
			dwMsgCount=0;
		}
		~HRecvPool()
		{
			sockList.clear();
		}
		HListener2 *	pThis;
		
		UNI_ThreadHandle hThread;
		UNI_DWORD		dwMsgCount;
		CHMutex			mutex;
		UNI_Semaphore	semaphore;
		list<CHTcpSock *> sockList;
	}RECVPOOL,*PRECVPOOL;
	PRECVPOOL	m_pRecvArray;
private:
	//插口处理线程,从自己的队列中取插口并处理
	static UNI_THREADPROC RecvThread(void * param)
	{
		CHTcpSock *pSock;
		PRECVPOOL pRecvPool=(PRECVPOOL)param;
		while (pRecvPool->pThis->m_bStop==UNI_FALSE)
		{
			UNI_WaitSemaphore(pRecvPool->semaphore);
			pSock=UNI_NULL;

			//从SOCK队列中先择一个
			pRecvPool->mutex.Lock();
			//////////////////////////////
			if(pRecvPool->dwMsgCount!=0)
			{
				pSock=(*pRecvPool->sockList.begin());
				pRecvPool->sockList.erase(pRecvPool->sockList.begin());
				pRecvPool->dwMsgCount--;
			}
			//////////////////////////////
			pRecvPool->mutex.UnLock();
			
			if(pSock)
			{
				pRecvPool->pThis->m_pUserFunc(pSock,pRecvPool->pThis->m_pUserParam);
				pSock->Close();
				delete pSock;
			}
			
		}
		UNI_ExitThread(0);
	}
	//插口接收线程线程,它把收到的插口放到处理线程的队列中
	static UNI_THREADPROC ListenThread(void * param)
	{
		HListener2 *pThis;
		CHTcpSock *pSock;
		UNI_DWORD dwCount,i;
		struct timeval time;
		fd_set read_set;
		pThis=(HListener2 *)param;

		dwCount=0;
		while (pThis->m_bStop==UNI_FALSE)
		{
			FD_ZERO(&read_set);
			pThis->m_Sock.AddToSet(&read_set);
			time.tv_sec=0;
			time.tv_usec=1000;
			if(1==select(0,&read_set,UNI_NULL,UNI_NULL,&time))
			{
				i=dwCount % pThis->m_RecvThreadCount;
				dwCount++;		
				
//				加入新收到的接口
				if(pSock=new CHTcpSock)
				{
					if(pSock->AcceptConnect(&pThis->m_Sock))
					{
						pThis->m_pRecvArray[i].mutex.Lock();
						////////////////////////////////////////////
						pThis->m_pRecvArray[i].sockList.insert (pThis->m_pRecvArray[i].sockList.end(), pSock);
						pThis->m_pRecvArray[i].dwMsgCount++;
						UNI_ReleaseSemaphore(pThis->m_pRecvArray[i].semaphore);
						//////////////////////////////////////////////
						pThis->m_pRecvArray[i].mutex.UnLock();
					}
					else
					{
						delete pSock;
					}
				}
				
				
			}
			
		}
		UNI_ExitThread(0);
	}
public:
	HListener2()
	{
		m_bStop=UNI_TRUE;
	}
	virtual ~HListener2()
	{
		End();
	}

	void End()
	{
		UNI_DWORD i;
		if(m_bStop==UNI_FALSE)
		{
			m_bStop=UNI_TRUE;
			m_Sock.Close();
	
			UNI_WaitCloseThread(m_hListenThread);
			if(m_pRecvArray)
			{
				for(i=0; i < m_RecvThreadCount; i++)
				{
					UNI_ReleaseSemaphore(m_pRecvArray[i].semaphore);
					UNI_WaitCloseThread(m_pRecvArray[i].hThread);
					UNI_CloseSemaphore(m_pRecvArray[i].semaphore);
					
				}
				delete [] m_pRecvArray;
				m_pRecvArray=UNI_NULL;
			}
			
		}
	
	}
	UNI_BOOL Start(UNI_DWORD dwThreadCount,UNI_IN HLISTENERCALLBACK pCall,UNI_IN void *pVoid,UNI_IN UNI_WORD wPort)
	{
		UNI_DWORD i;
		UNI_ThreadID     dwId;
		m_RecvThreadCount=0;
		this->m_pUserFunc=pCall;
		this->m_pUserParam=pVoid;
		if(m_Sock.ListenAt(wPort,100))
		if(m_pRecvArray= new RECVPOOL[dwThreadCount])
		{
			m_bStop=UNI_FALSE;
			for(i=0; i < dwThreadCount; i++)
			{
				m_pRecvArray[i].pThis=this;
				if(!UNI_CreateSemaphore(&m_pRecvArray[i].semaphore,0))
					break;
				if(!UNI_CreateThread(RecvThread,&m_pRecvArray[i],&m_pRecvArray[i].hThread,&dwId))
				{
					UNI_CloseSemaphore(m_pRecvArray[i].semaphore);
					break;
				}
			}
			if(i==dwThreadCount)
			{
				m_RecvThreadCount=dwThreadCount;
				if(UNI_CreateThread(ListenThread,this,&m_hListenThread,&dwId))
					return UNI_TRUE;
			}
			m_RecvThreadCount=0;
			m_bStop=UNI_TRUE;
			m_Sock.Close();
			for(i--;i>=0;i--)
			{
				UNI_CloseSemaphore(m_pRecvArray[i].semaphore);	
				UNI_WaitCloseThread(m_pRecvArray[i].hThread);
			}
			delete [] m_pRecvArray;
			m_pRecvArray=UNI_NULL;
		}
		
		return UNI_FALSE;
	}
};

#endif // !defined(AFX_HLISTENER2_H__BF0D5A93_4173_4C8F_B20D_E03405CC4D7E__INCLUDED_)