liveresource.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 ".\liveresource.h"
#include "Communicator.h"
#include "BufferMgr.h"

namespace NPLayer1 {

LiveResource::LiveResource(Communicator*c ) : 
	BaseResource(c), A_currBlock(NULL), V_currBlock(NULL), tmpBlockData(NULL)
{
}

LiveResource::~LiveResource(void) {
	Uninit(false);
}

// 第一次开始下载的初始化
P2P_RETURN_TYPE LiveResource::Init(
	string resname,				// 资源名
	string hashcode,			// 资源的Hash码
	string strSPList,			// SuperPeer列表的IP地址
	float bitRate)				// 频道的码率
{
	CriticalSection::Owner lock(dataLocker);
	try {
		// 不能重复初始化
		if(bInited)
			throw PRT_DUP_INIT;

		this->resname = resname;
		this->hashcode = hashcode;
		bufferMgr = &comm->g_bufferMgr;

		// 直播资源未知大小,需要的磁盘空间是小于缓冲文件大小的某个固定值
		// DEFAULT_SPACE必须保证整除BLOCK_SIZE
		blockMapSize = DEFAULT_SPACE/BLOCK_SIZE;

		// 初始化数组
		blockMapArray = new UINT[blockMapSize];
		blockIDArray = new UINT[blockMapSize];
		if(!blockMapArray || !blockIDArray)
			throw PRT_ALLOC;

		// 初始化数组所有成员为UINT_MAX
		fill(blockMapArray, blockMapArray+blockMapSize, UINT_MAX);
		fill(blockIDArray, blockIDArray+blockMapSize, UINT_MAX);

		P2P_RETURN_TYPE ret = ParseSPList(strSPList);
		if(ret != PRT_OK)
			throw ret;

		ret = InitBroadcastMember(bitRate);
		if(ret < PRT_OK)
			throw ret;
	}
	catch (P2P_RETURN_TYPE ret) {
		comm->logFile.StatusOut("Init Res Error!");
		Uninit(true);
		return ret;
	}

	bInited = TRUE;
	return PRT_OK;
}

void LiveResource::Uninit(bool force) {
	CriticalSection::Owner lock(dataLocker);
	if(bInited || force) {
		delete [] A_currBlock;
		delete [] V_currBlock;
		delete [] tmpBlockData;
		A_currBlock = V_currBlock = NULL;
		tmpBlockData = NULL;

		if(blockMapArray) {
			for(UINT i = 0; i < blockMapSize; ++i) {
				if(blockMapArray[i] != UINT_MAX)
					comm->g_bufferMgr.EraseIndex(blockMapArray[i]);
			}
		}

		UninitEx();
		bInited = FALSE;
	}
}

P2P_RETURN_TYPE LiveResource::InitBroadcastMember(float bitRate) {
	this->bitRate = bitRate;	// 码率
	memset(&spUpdate, 0, sizeof(SPUpdate));

	// 当前视频/音频块的数据
	if(!A_currBlock)
		A_currBlock = new BYTE[BLOCK_SIZE];
	if(!V_currBlock)
		V_currBlock = new BYTE[BLOCK_SIZE];
	if(!tmpBlockData)
		tmpBlockData = new BYTE[BLOCK_SIZE];
	if(!A_currBlock || !V_currBlock || !tmpBlockData)
		return PRT_ALLOC;
	A_currBlockID = V_currBlockID = UINT_MAX;
	A_leftDataInCurrBlock = V_leftDataInCurrBlock = 0;
	A_lastSampleTime = V_lastSampleTime = 0;
	A_last_OriginTime = V_last_OriginTime = 0;
	A_lastGetSampleFailedTime = V_lastGetSampleFailedTime = 0;
	programStartTime = 0;
	programState = PS_BOTH_BEGINS;
	totalBufferCount = 0;
	totalBufferTime  = 0;

	mediaArray.Clear();
	V_sentArray.Clear();
	A_sentArray.Clear();

	// 至少缓冲10块才能开始播放
	blocks4Play = max(static_cast<UINT>(TIME_FOR_PLAY*bitRate*1024/BLOCK_SIZE), 10);

	comm->logFile.StatusOut("blocks4play = %d. ", blocks4Play);

	return PRT_OK;
}

P2P_RETURN_TYPE LiveResource::LoadSample(
	SampleHeader& header,	// Sample头
	PBYTE& sampleData,		// Sample数据
	UINT sampleOff,			// 已经读取的长度
	const UINT maxSize,		// Sample最大长度
	const bool isAudio,		// 视频还是音频
	bool& bMediaTypeSample	// 是否媒体类型的特殊Sample
	)
{
	UINT& currBlockID = isAudio ? A_currBlockID : V_currBlockID;
	BYTE *currBlock = isAudio ? A_currBlock : V_currBlock;
	UINT& leftDataInCurrBlock = isAudio ? A_leftDataInCurrBlock : V_leftDataInCurrBlock;
	MediaArray& sentArray = isAudio ? A_sentArray: V_sentArray;

	bMediaTypeSample = false;

	// 尚未设置开始播放的位置
	if(currBlockID == UINT_MAX)
		return PRT_PROGRAM_SYNC;

	while(0 == leftDataInCurrBlock) { // not data in current block, load next
		// 读取当前块数据
		UINT bSize = 0;
		P2P_RETURN_TYPE ret = GetBlock(currBlockID, bSize, currBlock);
		if(ret < PRT_OK)
			return ret;
		assert(bSize <= BLOCK_SIZE);

		// 如果此块的媒体类型还没有提交给上层，那么首先返回媒体类型
		if(!sentArray.FindBlock(currBlockID)) {
			MediaInterval mInterval;
			if(!mediaArray.GetInterval(currBlockID, mInterval)) {
				// 还没有下载到当前块的媒体类型
				return PRT_PROGRAM_SYNC;
			}

			comm->logFile.StatusOut("Post %s Media Type[%d, %d)) @ %d.", 
				isAudio?"Audio":"Video", mInterval.start, mInterval.start+mInterval.size, currBlockID);

			// 在sample中填入媒体类型数据
			header.start = header.length = 0;
			header.bSyncPoint = 1;
			header.bAudioSample = isAudio;
			if(isAudio) {
				memcpy(sampleData, &mInterval.audioType, sizeof(mInterval.audioType));
				memcpy(sampleData+sizeof(mInterval.audioType), mInterval.audioData, mInterval.audioType.cbFormat);
				header.size = sizeof(mInterval.audioType) + mInterval.audioType.cbFormat;
			}
			else {
				memcpy(sampleData, &mInterval.videoType, sizeof(mInterval.videoType));
				memcpy(sampleData+sizeof(mInterval.videoType), mInterval.videoData, mInterval.videoType.cbFormat);
				header.size = sizeof(mInterval.videoType) + mInterval.videoType.cbFormat;
			}
			assert(header.size >= sizeof(MediaType));

			if(!sentArray.AddInterval(mInterval))
				assert(0);

			// video or audio stopped
			if(isAudio && mInterval.audioType.cbFormat == 0)
				A_currBlockID = UINT_MAX;
			if(!isAudio && mInterval.videoType.cbFormat == 0)
				V_currBlockID = UINT_MAX;

			bMediaTypeSample = true;
			return PRT_OK;
		}

		currBlockID++;

		// 此Block从完整的Sample开始, 或许是新的节目开始了，确认一下
		if(IsFirstBlockOfNewProgram(currBlock)) {
			// 获取此块对应的编码格式，如果暂时没有，则返回同步中的消息
			MediaInterval mInterval;
			if(!mediaArray.GetInterval(currBlockID-1, mInterval)) {
				currBlockID--;
				assert(0);
				return PRT_PROGRAM_SYNC;
			}

			comm->logFile.StatusOut("New Program %d!", currBlockID-1);
            /*
            if(!mInterval.audioType.bThisPinOnly && !mInterval.videoType.bThisPinOnly) {
				// 有音频和视频，准备同步过程
				if(programState == PS_BOTH_BEGINS)
					programState = (isAudio?PS_AUDIO_BEGINS:PS_VIDEO_BEGINS); // 进入同步阶段
				else if(isAudio) {
					if(programState == PS_VIDEO_BEGINS)
						programState = PS_BOTH_BEGINS;	// 发现视频已经在等待音频了，那么同步结束，开始播放
					else
						assert(0);	// 音频正在同步中，不可能又读取了新的Block
				}
				else {
					if(programState == PS_AUDIO_BEGINS)	// 发现音频已经在等待视频了，那么同步结束，开始播放
						programState = PS_BOTH_BEGINS;
					else
						assert(0);	// 视频正在同步中，不可能有读取了新的Block
				}
			}
			if(programState == PS_BOTH_BEGINS) {
				// 同步结束，选择节目开始时间
				if(mInterval.videoType.bThisPinOnly) {
					programStartTime = V_lastSampleTime+TIME_BETWEEN_PROGRAM; // 单路视频
				}
				// 有视频/音频两路的情况下,选择音频作为新节目开始的时间.
				comm->logFile.StatusOut("Program Start Old %I64d.", programStartTime);
				programStartTime = A_lastSampleTime+TIME_BETWEEN_PROGRAM;
				comm->logFile.StatusOut("Program Start New %I64d.", programStartTime);
			}
            */

			// 移动到第一个真正的Sample（非节目开头的标志Sample）的位置
			sampleOff = 0;
			leftDataInCurrBlock = BLOCK_SIZE-sizeof(UINT)*2-sizeof(SampleHeader);
			break;
		}

		// 第4~8个字节是first sample offset
		UINT sampleOffset = *((UINT*)currBlock+1);
		if(sampleOffset != UINT_MAX && (sampleOffset < sizeof(UINT)*2 || sampleOffset > BLOCK_SIZE)) {
			assert(0);

			// 删除这个错误的Block，并且跳过这个Block
			DelBlock(currBlockID-1);
			leftDataInCurrBlock = 0;
			sampleOff = 0;
			continue; // 错误的SampleOffset，抛弃当前块
		}

		// TODO: 由于CaptureServer的一个BUG，这里作一个修正，随着所有CaptureServer的更新，这里可以去除。 2005.04.18
		if(sampleOffset == BLOCK_SIZE)
			sampleOffset = UINT_MAX;

		if(sampleOff == 0) { // 开始读取新的Sample
			if(sampleOffset == UINT_MAX)
				continue; // 新的Sample不在当前Block中，继续寻找下一个
			else 
				leftDataInCurrBlock = BLOCK_SIZE-sampleOffset; // 新的Sample位置
		}
		else { // 继续读取剩余的数据
			leftDataInCurrBlock = BLOCK_SIZE-sizeof(UINT)*2; 
			if(!(sampleOffset==UINT_MAX || sampleOff < sizeof(SampleHeader) || sampleOffset == sizeof(UINT)*2+header.size-sampleOff)) {
				char* temp = NULL;
				MD5 md5(reinterpret_cast<BYTE*>(currBlock), BLOCK_SIZE);
				temp = md5.hex_digest();
				comm->logFile.StatusOut("Bad Block %d(%s).", currBlockID-1, temp);
				delete [] temp;

				assert(0);
				// 删除这个错误的Block，并且跳过这个Block
				DelBlock(currBlockID-1);
				leftDataInCurrBlock = 0;
				sampleOff = 0;
				continue;
			}
		}
		assert(leftDataInCurrBlock <= BLOCK_SIZE);
		break;
	}

	if(programState != PS_BOTH_BEGINS) { // 正在同步,应该可以投出类型数据
		if(!isAudio) {
			if(programState == PS_VIDEO_BEGINS)
				return PRT_PROGRAM_SYNC;	// 视频正在等待同步，暂时不返回数据
			//else
				// 音频正在同步，不影响视频，所以继续
		}
		else {
			if(programState == PS_AUDIO_BEGINS)
				return PRT_PROGRAM_SYNC;	// 音频正在等待同步，暂时不返回数据
			//else
				// 视频正在同步，不影响音频，所以继续
		}
	}

	BYTE* startOff = currBlock+(BLOCK_SIZE - leftDataInCurrBlock);

	// 尝试读取SampleHeader，直到读取了sizeof(SampleHeader)个字节，即dataOff == sizeof(SampleHeader)
	if(sampleOff < sizeof(SampleHeader)) {
		if(leftDataInCurrBlock < sizeof(SampleHeader)-sampleOff) {
			// 数据仍然不足sizeof(SampleHeader)，继续读取下一个块
			memcpy((BYTE*)&header + sampleOff, startOff, leftDataInCurrBlock);
			sampleOff += leftDataInCurrBlock;
			leftDataInCurrBlock = 0;
			return LoadSample(header, sampleData, sampleOff, maxSize, isAudio, bMediaTypeSample);
		}
		else {
			// 数据已经足够，读取完整的sample header
			memcpy((BYTE*)&header + sampleOff, startOff, sizeof(SampleHeader)-sampleOff);
			assert(header.size >= sizeof(SampleHeader) && header.size <= maxSize);
			if(header.size < sizeof(SampleHeader) || header.size > maxSize) {
				// 错误的Header，抛弃当前块，重新读取新的Sample
				sampleOff = 0;
				leftDataInCurrBlock = 0;
				return LoadSample(header, sampleData, sampleOff, maxSize, isAudio, bMediaTypeSample);
			}
			leftDataInCurrBlock -= (sizeof(SampleHeader)-sampleOff);
			sampleOff = sizeof(SampleHeader);
		}
	}

	startOff = currBlock+(BLOCK_SIZE - leftDataInCurrBlock);
	// 读取SampleHeader之外的数据
	if(sampleOff >= sizeof(SampleHeader)) {
		if(leftDataInCurrBlock < header.size-sampleOff) {
			// 当前块中剩下的数据不足，继续读取下一块
			if(sampleData)
				memcpy(sampleData+sampleOff-sizeof(SampleHeader), startOff, leftDataInCurrBlock);
			sampleOff += leftDataInCurrBlock;
			leftDataInCurrBlock = 0;
			return LoadSample(header, sampleData, sampleOff, maxSize, isAudio, bMediaTypeSample);
		}
		else {
			// 读取到完整的Sample
			if(sampleData)
				memcpy(sampleData+sampleOff-sizeof(SampleHeader), startOff, header.size-sampleOff);
			leftDataInCurrBlock -= (header.size-sampleOff);
		}
	}

	assert(leftDataInCurrBlock <= BLOCK_SIZE);
	return PRT_OK;
}

P2P_RETURN_TYPE LiveResource::GetData(
									  SampleHeader& header,	// out, 数据头
									  PBYTE& pData,			// out, 存储数据的缓冲区
									  const UINT maxSize,	// in, 缓冲区的长度
									  const bool bAudio,	// in, 获取音频还是视频
									  const bool bKeySample	// in, 是否寻找关键帧
									  )
{
	if(!bInited) // not started
		return PRT_NOT_INIT;
	CriticalSection::Owner lock(dataLocker);

	bool bMediaTypeSample = false;

	for(;;) {
		for(;;) {
			P2P_RETURN_TYPE ret = LoadSample(header, pData, 0, maxSize, bAudio, bMediaTypeSample);
			if(ret < PRT_OK) {