networkstub.cpp

mysee网络直播源代码Mysee Lite是Mysee独立研发的网络视频流媒体播放系统。在应有了P2P技术和一系列先进流媒体技术之后

/*
 *  Openmysee
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */
#include "stdafx.h"
#include "NetWorkStub.h"
#include "ChannelMgr.h"

#include <string>
using namespace std;

#ifdef USE_LOG
FILE* log = NULL;
#pragma message("will generate log for networkstub")
#endif

NetworkStub::NetworkStub()
: mainThread(NULL), sock(NULL), chnnMgr(NULL) //log("#NetworkStub")
{
}

NetworkStub::~NetworkStub()
{
	if(mainThread)
	{
		closesocket(sock);
		if( WaitForSingleObject(mainThread, 3000) != WAIT_OBJECT_0 )
		{
			// 如果超过3秒还没退出，再Terminate
			TerminateThread(mainThread, 1L);
		}
		CloseHandle(mainThread);
	}

#ifdef USE_LOG
	if(log) fclose(log);
#endif
}

// 初始化,并启动新线程处理socket消息
bool NetworkStub::Init( SOCKET sockFromProxy, CChannelMgr* chnnMgr )
{
	unsigned int threadID;

	this->sock = sockFromProxy;
	this->chnnMgr = chnnMgr;
	mainThread = (HANDLE) _beginthreadex(NULL, 0, &_ThreadFunc, this, 0, &threadID);
	if( ! mainThread )
		return false;

	return true;
}

// Thread function
unsigned int CALLBACK NetworkStub::_ThreadFunc(void* arg)
{	
	NetworkStub* pThis = (NetworkStub*)arg;

	// data for getData
	PBYTE pData = NULL;

	// Run
	pThis->_ProcessProxy(pData);

	//chnnMgr->Stop() should be called.
	if(pThis->chnnMgr)
		pThis->chnnMgr->Stop();

	//release resources
	if(pData)
		free(pData);
	return 0;
}

void NetworkStub::_ProcessProxy(PBYTE& pData)
{
	// tcp client
	TCPClient proxyTcp(sock);

	// first 6 bytes of each msg
	UINT8 msgType, subMsgType;
	UINT length;
	PBYTE data;

	// current length of pData
	UINT curDataLen = 0;

	// main loop: parse message.
	// return immdiately if error occur or receive stop message.
	PSTREAM s = proxyTcp.InitData(1024);
	while( s = proxyTcp.Recv(NULL, 6) )
	{
		in_uint8(s, msgType);
		in_uint8(s, subMsgType);
		in_uint32_be(s, length);

		switch(msgType){ //response to each type
		case MSG_REQUEST:
			{
				UINT8 retvalue = 0;
				if( length > 0 )
				{
					s = proxyTcp.Recv(s, length);
					if(!s)
						return;
					in_uint8p(s, data, length);

					retvalue = chnnMgr->Request((LPCTSTR)data) ? 1 : 0;
				}

#ifdef USE_LOG
				if( !log )
					log = fopen("TestStub.txt", "w+");
				fprintf(log, "recv request: %s, %d\n", (LPCTSTR)data, length);
#endif

				if( ! proxyTcp.SendEx(MSG_REQUEST, retvalue, NULL, 0) )
					return;

#ifdef USE_LOG
				fprintf(log, "send response: type %d\n\n", MSG_REQUEST);
#endif
			}
			break;

		case MSG_SENDMSGTOSHOW:
			{
				in_uint8p(s, data, length); 

#ifdef USE_LOG
				fprintf(log, "recv sendmsgtoshow: %s\n", (LPCTSTR)data);
#endif

				chnnMgr->SendMsgToShow((LPCTSTR)data, length);

				if( ! proxyTcp.SendEx(MSG_SENDMSGTOSHOW, 0, NULL, length) )
					return;

#ifdef USE_LOG
				fprintf(log, "send response: type %d\n\n", MSG_SENDMSGTOSHOW);
#endif
			}
			break;

		case MSG_STOP:
			{
#ifdef USE_LOG
				fprintf(log, "recv stop.\n\n");
#endif
			}
			return; // return if receive stop msg. chnnMgr->Stop() will be called later

		case MSG_GETDATA:
			{
				bool needGet = true; // whether need to GetData from chnnMgr

				if( length != sizeof(UINT) )
					needGet = false; // error

				bool bAudio = (subMsgType & 0x2) ? true : false;
				bool bKeySample = (subMsgType & 0x1) ? true : false;

				s = proxyTcp.Recv(s, sizeof(UINT));
				if(!s)
					return;
				in_uint8p(s, data, sizeof(UINT));
				UINT maxSize = *((UINT*)data);
				if( maxSize==0 ) 
					needGet = false;

#ifdef USE_LOG
				fprintf(log, "recv getdata: bAudio %d, bKeySample %d, maxSize:%u\n", bAudio?1:0, bKeySample?1:0, maxSize);
#endif

				if( !pData || curDataLen<maxSize )
				{
					curDataLen = maxSize;
					pData = (PBYTE)realloc(pData, maxSize);
					if( !pData )
						needGet = false;
				}

				int ret = -1; 
				SampleHeader header;
				if( needGet )
					ret = chnnMgr->GetData(header, pData, maxSize, bAudio, bKeySample);

				// send response to proxy
				UINT8 retValue; // subMsgType of response to proxy 
				if( ret > 0 ) // 成功
				{
					retValue = 1;
					if( ! proxyTcp.SendEx2(MSG_GETDATA, retValue, (PBYTE)&header, sizeof(SampleHeader), pData, header.size ) )
						return;

#ifdef USE_LOG
					fprintf(log, "send response: type %d, subType %u, headerlen %u, datalen %u\n", MSG_GETDATA, retValue, sizeof(SampleHeader), header.size);
					// header
					fprintf(log, "header: %u, %u, %u, %u, %u, %u, %I64d\n", header.size, header.bDiscontinuity, header.bPreroll, header.bSyncPoint, header.bAudioSample, header.length, header.start);
					// data
					UINT i=0; string datastr = "data: "; char tmp[16];
					for( i=0; i<header.size; i++){
						itoa(pData[i], tmp, 10);
						datastr.append( tmp );
						datastr.append( " " );
					}
					fprintf(log, "%s\n\n", datastr.data() );
#endif
				}
				else if( ret==0 ) //暂时没有数据
				{
					retValue = 0;
					if( ! proxyTcp.SendEx2(MSG_GETDATA, retValue, NULL, 0, NULL, 0) )
						return;

#ifdef USE_LOG
					fprintf(log, "send response: type %d, subType %u\n", MSG_GETDATA, retValue);
#endif
				}
				else // 错误
				{
					retValue = -1;
					if( ! proxyTcp.SendEx2(MSG_GETDATA, retValue, NULL, 0, NULL, 0) )
//						proxyTcp.SendEx2(MSG_GETDATA, retValue, NULL, 0, NULL, 0);
					return;

#ifdef USE_LOG
					fprintf(log, "send response: type %d, subType %u\n", MSG_GETDATA, retValue);
#endif
				}

			}
			break;
		}
	}
}