clientsocket.cpp

一个多线程的网络数据采集系统（客户端）

// ClientSocket.cpp : 实现文件
//

#include "stdafx.h"
#include "WspProcess.h"
#include "ClientSocket.h"
#include "ClientThread.h"


/////////////////////////////////////////////////////////////////////////////
// CPkg

IMPLEMENT_DYNCREATE(CPkg, CObject)
/**
 *@function 获取名称
 */
CPkg::CPkg()
{
	Init();
}

CPkg::CPkg(CPkg& pkg)
{
	MsgFlag = pkg.MsgFlag;
	datatype = pkg.datatype;
	int i = datatype;
	switch(i)
	{
        case TYPE_CHAR:
			data.d_char = pkg.data.d_char;break;
		case TYPE_INT:
			data.d_int = pkg.data.d_int;break;
		case TYPE_LONG:
			data.d_long = pkg.data.d_long;break;
		case TYPE_FLOAT:
			data.d_float = pkg.data.d_float;break;
		case TYPE_DOUBLE:
			data.d_double = pkg.data.d_double;break;
		default:
			break;
	}
	uComNo = pkg.uComNo;
}

CPkg::~CPkg()
{
}

/////////////////////////////////////////////////////////////////////////////
// CPkg diagnostics

#ifdef _DEBUG
void CPkg::AssertValid() const
{
	CObject::AssertValid();
}

void CPkg::Dump(CDumpContext& dc) const
{
	CObject::Dump(dc);
}
#endif //_DEBUG

/////////////////////////////////////////////////////////////////////////////
// CPkg serialization

void CPkg::Serialize(CArchive& ar)
{
	if (ar.IsStoring())
	{
		ar << MsgFlag;
//		int dt = datatype;
//        ar << dt;
		switch(datatype){
			case TYPE_CHAR:
				ar << data.d_char;break;
			case TYPE_INT:
				ar << data.d_int;break;
			case TYPE_LONG:
				ar << data.d_long;break;
			case TYPE_FLOAT:
				ar << data.d_float;break;
			case TYPE_DOUBLE:
				ar << data.d_double;break;
			default:
				break;
		}
        ar << uComNo;
	}
	else
	{
		ar >> MsgFlag;
/*		int dt ;
        ar >> dt;
		switch(dt){
			case TYPE_CHAR:
				dt = TYPE_CHAR; ar >> data.d_char;break;
			case TYPE_INT:
				dt = TYPE_INT; ar >> data.d_int;break;
			case TYPE_LONG:
				dt = TYPE_LONG; ar >> data.d_long;break;
			case TYPE_FLOAT:
				dt = TYPE_FLOAT;*/ ar >> data.d_float;/*break;
			case TYPE_DOUBLE:
				dt = TYPE_DOUBLE; ar >> data.d_double;break;
			default:
				break;
		}*/
        ar >> uComNo;
	}
}

void CPkg::Init()
{
	MsgFlag = UNDEFINED;
}

//////////////////////////////////////////////////////////////////////
// CPkgQueue
///////////////
// Add Message
//////////////
void CPkgQueue::Add(CPkg *pMsg)
{
	CSingleLock slock(&m_mutex);

	if (slock.Lock(1000))  // timeout in milliseconds, default= INFINITE
	{
		AddTail(pMsg); // fifo
	}
}

//////////////////
// Remove Message
//////////////////
CPkg* CPkgQueue::Remove()
{
	CSingleLock slock(&m_mutex);

	if (slock.Lock(1000))  // timeout in milliseconds, default= INFINITE
	{
		if (!IsEmpty())
			return (CPkg*)RemoveHead();
	}
	return NULL;
}

//////////////////////////////////////////////////////////////////////
// CClientSocket

CClientSocket::CClientSocket(CClientThread *pThread)
{
	m_pOwner = pThread;
	m_pSocketFile = NULL;
	m_pArchiveIn = NULL;
	m_pArchiveOut = NULL;
	m_bReceive = false;
	m_bClose = true;
}

CClientSocket::~CClientSocket()
{
	Finalize();
}

// CClientSocket 成员函数
void CClientSocket::Init()
{
	m_pSocketFile = new CSocketFile(this);
	m_pArchiveIn = new CArchive(m_pSocketFile, CArchive::load);
	m_pArchiveOut = new CArchive(m_pSocketFile, CArchive::store);
}

BOOL CClientSocket::Open(const CString& sConnStr)
{
	if(! Create())
	{
		AfxMessageBox(_T("failing to Create Client Socket!"));
		return FALSE;
	}
	
	if(! Connect(sConnStr,1500)) //请求与服务器端建立连接
	{	
		AfxMessageBox("Not Connected");
		return FALSE;
	}
	Init();
	m_bClose = false;
	return TRUE;
}

void CClientSocket::ReceivePkg(CPkg *pkg)
{
	ASSERT(pkg);
	if(m_pArchiveIn == NULL)return;
	try{
		do{
			pkg->Serialize(*m_pArchiveIn);
		}while(!m_pArchiveIn->IsBufferEmpty());
	}
	catch(CFileException*){
		m_pArchiveIn->Abort();
	}
}

void CClientSocket::SendPkg(CPkg *pkg)
{
	ASSERT(pkg);
	if(m_pArchiveOut == NULL)return;
	try{
		pkg->Serialize(*m_pArchiveOut);
		m_pArchiveOut->Flush();
	}
	catch(CFileException*){
		m_pArchiveOut->Abort();
	}
}

void CClientSocket::Finalize()
{
	if(m_pArchiveIn != NULL){
		delete m_pArchiveIn;
		m_pArchiveIn = NULL;
	}
	if(m_pArchiveOut != NULL){
		delete m_pArchiveOut;
		m_pArchiveOut = NULL;
	}
	if(m_pSocketFile != NULL){
		delete m_pSocketFile;
		m_pSocketFile = NULL;
	}
	Close();
}

void CClientSocket::OnReceive(int nErrorCode)
{
	// TODO: 在此添加专用代码和/或调用基类
	CSocket::OnReceive(nErrorCode);
	m_bReceive = true;
	m_pOwner->ProcessSocket();
}


void CClientSocket::OnClose(int nErrorCode)
{
	CSocket::OnClose(nErrorCode);
	m_bClose = true;
	m_pOwner->ProcessSocket();
}