cworkersocket.cpp

《DirectShow开发指南》配套代码代码,需要DirectX SDK 7以上支持。

// 
// CWorkerSocket.cpp
// 

#include "stdafx.h"
#include "CWorkerSocket.h"

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

//////////////////////////////////////////////////////////////////////////////
CWorkerSocket::CWorkerSocket()
{
	m_hSocket      = NULL;
	m_bReceiving   = false;
	m_bSending     = false;

	// Initialize thread event
	m_hThrdFinish = ::CreateEvent(NULL,    // pointer to security attributes 
	    						TRUE,      // flag for manual-reset event 
								FALSE,     // flag for initial state 
								NULL);     // pointer to event-object name
	m_hSendFinish = ::CreateEvent(NULL, 
	    						TRUE, 
								FALSE, 
								NULL); 
}

CWorkerSocket::~CWorkerSocket()
{
	// Close the socket
	Disconnect();

	// Release the thread event
	if (m_hThrdFinish)
	{
		CloseHandle(m_hThrdFinish);
		m_hThrdFinish = NULL;
	}
	if (m_hSendFinish)
	{
		CloseHandle(m_hSendFinish);
		m_hSendFinish = NULL;
	}
}

// Attach a connected sockt to this object
bool CWorkerSocket::Attach(SOCKET inSock)
{
	if (m_hSocket == NULL)
	{
		m_hSocket = inSock;
		return true;
	}
	return false;
}

SOCKET CWorkerSocket::Detach(void)
{
	SOCKET hSock = m_hSocket;
	m_hSocket    = NULL;
	return hSock;
}

// Send data on connected sockets, blockingly
int CWorkerSocket::Send(char * inData, int inLen)
{
	if (m_hSocket)
	{
		char * pData = inData; //new char[inLen];
		if (!pData)
			return E_SOCKET_FAIL;

		TIMEVAL tm;
		fd_set wfds;
		FD_ZERO(&wfds);
		FD_SET(m_hSocket, &wfds);
		tm.tv_sec  = 1;
		tm.tv_usec = 0;
		// Make sure the current socket can write
		if (select(0, NULL, &wfds, NULL, &tm))
		{
			int nret   = 0;
			int nCount = inLen;
			do 
			{
				nret = send(m_hSocket, pData, nCount, 0);
				if (nret == SOCKET_ERROR)
				{
					return E_SOCKET_FAIL;
				}
				nCount -= nret;
				pData  += nret;
				if (nCount > 0) 
					Sleep(20);
			} while (nCount > 0);
			return S_SOCKET_OK;
		}
		else
		{
			// The socket cannot write currently
			return E_SOCKET_NOT_READY;
		}
	}

	return E_SOCKET_FAIL;
}

int CWorkerSocket::Receive(char * outBuf, int inLen)
{
	if (m_hSocket)
	{
		// Check socket for readability 
		TIMEVAL tv;
		tv.tv_sec  = 0;
		tv.tv_usec = 0;
		fd_set rfds;
		FD_ZERO(&rfds);
		FD_SET(m_hSocket, &rfds);
		if (select(1, &rfds, NULL, NULL,&tv))
		{
			int    nCount = inLen;
			int    lenret = 0;
			char * pBuf   = outBuf;
			do 
			{
				lenret = recv(m_hSocket, pBuf, nCount, 0);
				if (lenret == SOCKET_ERROR) 
				{
					return E_SOCKET_FAIL;
				}
				else if (lenret == 0)
				{
					// Indicate the socket disconnected
					return E_SOCKET_CLOSE;
				}
				nCount -= lenret;
				pBuf   += lenret;
				if (nCount > 0) 
					Sleep(20);
			} while (nCount > 0);
			return S_SOCKET_OK;
		}
		else
		{
			// The socket not ready to read
			return E_SOCKET_NOT_READY;
		}
	}
	return E_SOCKET_FAIL;
}

void CWorkerSocket::Disconnect(void)
{
	if (m_hSocket)
	{
		// Terminate receiving and sending thread if necessary
		if (m_bReceiving)
			StopReceiving();
		if (m_bSending)
			StopSending();

		closesocket(m_hSocket);
		m_hSocket = NULL;
	}
}

// Connect to the remote socket
bool CWorkerSocket::Connect(char * inTarget, int inPort)
{
	Disconnect();

	// Create a new socket to connection
	m_hSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (m_hSocket == INVALID_SOCKET)
	{
		m_hSocket = NULL;
		return false;
	}
	BOOL sopt = TRUE;
	setsockopt(m_hSocket, IPPROTO_TCP, TCP_NODELAY, (char *)&sopt, sizeof(BOOL));
	setsockopt(m_hSocket, SOL_SOCKET, SO_DONTLINGER, (char *)&sopt, sizeof(BOOL));

	// Resolve hostname
	DWORD  addr = inet_addr(inTarget);
	if (addr == INADDR_NONE) 
	{
		struct hostent *he = gethostbyname(inTarget);
		if (he == NULL) 
		{
			closesocket(m_hSocket);
			m_hSocket = NULL;
			return false;
		}
		addr = *(DWORD *)(he->h_addr_list[0]);
	}

	// Create socket address
	SOCKADDR_IN   saddr;
	memset(&saddr, 0, sizeof(SOCKADDR_IN));
	saddr.sin_addr.S_un.S_addr = addr;
	saddr.sin_family           = AF_INET;
	saddr.sin_port             = htons((WORD)inPort);
	// Connect to the remote host
	if (connect(m_hSocket, (SOCKADDR *)&saddr, sizeof(SOCKADDR_IN)) != 0) 
	{
		closesocket(m_hSocket);
		m_hSocket = NULL;
		return false;
	}
	return true;
}

////////////////////////////////////////////////////////////////////////////
// Start a data receiving thread
bool CWorkerSocket::StartReceiving(void)
{
	if (m_hSocket)
	{
		if (m_bReceiving)
			return true;

		m_bReceiving = true;
		ResetEvent(m_hThrdFinish);
		// Make socket blocking safely
		u_long   nonBlock = FALSE;
		ioctlsocket(m_hSocket, FIONBIO, &nonBlock);
		AfxBeginThread((AFX_THREADPROC)CWorkerSocket::ReceivingThrd, this);
		return true;
	}
	return false;
}

bool CWorkerSocket::StopReceiving(void)
{
	if (m_hSocket)
	{
		if (!m_bReceiving)
			return true;

		m_bReceiving = false;
		// Make socket nonblocking to terminate receiving thread
		char pData[sizeof(MSG_HEADER)];
		PMSG_HEADER pMsg = (PMSG_HEADER) pData;
		pMsg->nDataSize  = 0;
		pMsg->nMsgType   = RECV_EXIT_REQUEST;
		Send(pData, sizeof(MSG_HEADER));
	//	u_long   nonBlock = TRUE;
	//	ioctlsocket(m_hSocket, FIONBIO, &nonBlock);
		WaitForSingleObject(m_hThrdFinish, 2000);
		// Restore to blocking socket
	//	nonBlock = FALSE;
	//	ioctlsocket(m_hSocket, FIONBIO, &nonBlock);
		return true;
	}
	return false;
}

// Virtual method, receive loop
void CWorkerSocket::ReceivingLoop(void)
{
}

UINT CWorkerSocket::ReceivingThrd(void * pParam)
{
	CWorkerSocket * pSock = (CWorkerSocket *) pParam;
	if (pSock != NULL)
	{
		pSock->ReceivingLoop();
	}

	SetEvent(pSock->m_hThrdFinish);
	return 1;
}

////////////////////////////////////////////////////////////////////////////
bool CWorkerSocket::StartSending(void)
{
	if (m_hSocket)
	{
		if (m_bSending)
			return true;

		m_bSending = true;
		ResetEvent(m_hSendFinish);
		// Make socket blocking safely
	//	u_long   nonBlock = FALSE;
	//	ioctlsocket(m_hSocket, FIONBIO, &nonBlock);
		AfxBeginThread((AFX_THREADPROC)CWorkerSocket::SendingThrd, this);
		return true;
	}
	return false;
}

bool CWorkerSocket::StopSending(void)
{	
	if (m_hSocket)
	{
		if (!m_bSending)
			return true;

		m_bSending = false;
		// Make socket nonblocking to terminate sending thread
	//	u_long   nonBlock = TRUE;
	//	ioctlsocket(m_hSocket, FIONBIO, &nonBlock);
		WaitForSingleObject(m_hSendFinish, 2000);
		// Restore to blocking socket
	//	nonBlock = FALSE;
	//	ioctlsocket(m_hSocket, FIONBIO, &nonBlock);
		return true;
	}
	return false;
}

// Virtual method: data-sending loop
void CWorkerSocket::SendingLoop(void)
{
}

UINT CWorkerSocket::SendingThrd(void * pParam)
{	
	CWorkerSocket * pSock = (CWorkerSocket *) pParam;
	if (pSock != NULL)
	{
		pSock->SendingLoop();
	}

	SetEvent(pSock->m_hSendFinish);
	return 1;
}