cstreamsocket.cpp

陆其明的实务精选中附带光盘中的视频聊天源代码

// 
// CStreamSocket.cpp
// 

/*-----------------------------------------------------*\
			HQ Tech, Make Technology Easy!       
 More information, please go to http://hqtech.nease.net.
/*-----------------------------------------------------*/

#include "stdafx.h"
#include <streams.h>
#include "CStreamSocket.h"
#include "CFilterNetReceiver.h"
#include "GlobalDefs.h"

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

//////////////////////////////////////////////////////////////////////////////
CStreamSocket::CStreamSocket()
{
	mSocket      = INVALID_SOCKET;
	mIsConnected = FALSE;

	mIsReceiving = FALSE;
	mIsSending   = FALSE;
	mRcvThread   = NULL;
	mSndThread   = NULL;

	mOwnerFilter = NULL;
}

CStreamSocket::~CStreamSocket()
{
	Detach();
	StopSending();
	StopReceiving();
}

void CStreamSocket::SetOwnerFilter(CFilterNetReceiver * inFilter)
{
	mOwnerFilter = inFilter;
}

BOOL CStreamSocket::Attach(SOCKET inSocket)
{
	if (mSocket != INVALID_SOCKET)
	{
		return FALSE;
	}

	mSocket      = inSocket;
	mIsConnected = TRUE;
	return TRUE;
}

void CStreamSocket::Detach(void)
{
	if (mSocket != INVALID_SOCKET)
	{
		closesocket(mSocket);
		mSocket = INVALID_SOCKET;
		mIsConnected = FALSE;
	}
}

SOCKET CStreamSocket::GetSocket(void)
{
	SOCKET ret   = mSocket;
	mSocket      = INVALID_SOCKET;
	mIsConnected = FALSE;
	return ret;	
}

BOOL CStreamSocket::ConnectTo(DWORD inTargetIP, WORD inPort)
{
	if (mIsConnected)
	{
		return TRUE;
	}

	mSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (mSocket != INVALID_SOCKET)
	{
		BOOL sopt = TRUE;
		setsockopt(mSocket, IPPROTO_TCP, TCP_NODELAY, 
			(char *)&sopt, sizeof(BOOL));
		setsockopt(mSocket, SOL_SOCKET, SO_DONTLINGER, 
			(char *)&sopt, sizeof(BOOL));

		SOCKADDR_IN   saddr;
		memset(&saddr, 0, sizeof(SOCKADDR_IN));
		saddr.sin_addr.S_un.S_addr = htonl(inTargetIP);
		saddr.sin_family           = AF_INET;
		saddr.sin_port             = htons((WORD)inPort);
		if (connect(mSocket, (SOCKADDR *)&saddr, sizeof(SOCKADDR_IN)) != 0) 
		{
			Detach();
			return FALSE;
		}
		mIsConnected = TRUE;
		return TRUE;
	}
	return FALSE;
}

BOOL CStreamSocket::StartReceiving(void)
{
	if (mSocket != INVALID_SOCKET)
	{
		if (mIsReceiving)
		{
			return TRUE;
		}

		DWORD threadID = 0;
		mRcvThread = CreateThread(NULL, 0, ReceivingThrd, 
			this, 0, &threadID);
		return (mRcvThread != NULL);
	}
	return FALSE;
}

void CStreamSocket::StopReceiving(void)
{
	if (mIsReceiving)
	{
		Detach();
		if (mRcvThread != NULL) 
		{
            WaitForSingleObject(mRcvThread, INFINITE);
			mRcvThread = NULL;
		}
	}
}

BOOL CStreamSocket::StartSending(void)
{
	if (mSocket != INVALID_SOCKET)
	{
		if (mIsSending)
		{
			return TRUE;
		}

		DWORD threadID = 0;
		mSndThread = CreateThread(NULL, 0, SendingThrd, 
			this, 0, &threadID);
		return (mSndThread != NULL);
	}
	return FALSE;
}

void CStreamSocket::StopSending(void)
{
	if (mIsSending)
	{
		Detach();
		if (mSndThread != NULL) 
		{
            WaitForSingleObject(mSndThread, INFINITE);
			mSndThread = NULL;
		}
	}
}

DWORD WINAPI CStreamSocket::ReceivingThrd(void * pParam)
{
	CStreamSocket * pSock = (CStreamSocket *) pParam;
	if (pSock)
	{
		pSock->mIsReceiving = TRUE;
		pSock->ReceivingLoop();
		return 1;
	}	
	return 0;
}

DWORD WINAPI CStreamSocket::SendingThrd(void * pParam)
{
	CStreamSocket * pSock = (CStreamSocket *) pParam;
	if (pSock)
	{
		pSock->mIsSending = TRUE;
		pSock->SendingLoop();
		return 1;
	}	
	return 0;
}

// Used in the Net Receiver Filter
void CStreamSocket::ReceivingLoop(void)
{
	char   buf[1024];
	int    bytes = 0;
	Pack_Header * pHeader = (Pack_Header *) buf;
	PBYTE  pSampleBuffer;

	while (mIsReceiving && mOwnerFilter)
	{
		// Receive a pack header first
		bytes = recv(mSocket, buf, sizeof(Pack_Header), 0);
		if (bytes == SOCKET_ERROR || bytes == 0)
		{
			Detach();
			mIsReceiving = FALSE;
			break;
		}

		pHeader->my_ntoh();
		if (pHeader->pack_type == PT_AudioMediaType ||
			pHeader->pack_type == PT_VideoMediaType)
		{
			// Receive media type
			bytes = recv(mSocket, buf+sizeof(Pack_Header), pHeader->pack_size, 0);
			if (bytes == SOCKET_ERROR || bytes == 0)
			{
				Detach();
				mIsReceiving = FALSE;
				break;
			}
			// Notify the filter to build filter graph
			mOwnerFilter->SetupMediaType(pHeader->pack_type, buf+sizeof(Pack_Header), 
				pHeader->pack_size);
		}
		else
		{
			// Receive media data
			BOOL pass = FALSE;
			do
			{
				pass = mOwnerFilter->GetSampleBuffer(&pSampleBuffer);
				if (!pass)
				{
					Sleep(100);
				}
			} while (mIsReceiving && !pass);

			long required = pHeader->pack_size;
			long received = 0;
			while (pass && mIsReceiving && required > 0)
			{
				bytes = recv(mSocket, (char*)(pSampleBuffer+received), required, 0);
				if (bytes == SOCKET_ERROR || bytes == 0)
				{
					Detach();
					mIsReceiving = FALSE;
					break;
				}
				received += bytes;
				required -= bytes;	
			}
			// Deliver this sample
			mOwnerFilter->DeliverHoldingSample(pHeader->pack_size);
		//	Sleep(10);
		}
	}
}

void CStreamSocket::SendingLoop(void)
{
}