liveresource.cpp

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

				if(bAudio)
					A_lastGetSampleFailedTime = timeGetTime(); // 记录GetSample失败的时间
				else
					V_lastGetSampleFailedTime = timeGetTime(); // 记录GetSample失败的时间
				comm->logFile.StatusOut("GetData failed. %d", ret);
				return ret;
			}
			// 如果是目标类型，则完成寻找
			if(bAudio == header.bAudioSample)
				break;
		}

		// 是否正在寻找视频的关键Sample，如果是，当前Sample是目标Sample吗？
		if(!bKeySample || (!bAudio && header.bSyncPoint))
			break;
	}

	if(bAudio)
		A_lastGetSampleFailedTime = 0; // GetSample成功了
	else
		V_lastGetSampleFailedTime = 0; // GetSample成功了


	if(!bMediaTypeSample) {
		A_last_OriginTime = header.start+header.length;

		// 加上节目开始的时间
		header.start += programStartTime;
		
		//assert(header.start+1*10000000 >= (isAudio?A_lastSampleTime:V_lastSampleTime));
		//assert(header.bDiscontinuity == 0);
		header.bDiscontinuity = 0;

		// 记录最近一个Sample的时间
		if(bAudio)
			A_lastSampleTime = header.start;
		else
			V_lastSampleTime = header.start;

		comm->logFile.StatusOut("%s Sample Time %I64d(%I64d) at Block %d", bAudio?"Audio":"Video", header.start, 
			header.start - programStartTime, bAudio?A_currBlockID-1:V_currBlockID-1);
	}
	return PRT_OK;
}

// 存储Block，在存储之前检查此Block及其后面的一个Block是否正确
P2P_RETURN_TYPE LiveResource::PutBlock(UINT blockID, UINT blockSize, PBYTE data) {
	if(!bInited) // not started
		return PRT_NOT_INIT;
	CriticalSection::Owner lock(dataLocker);
    assert(mediaArray.FindBlock(blockID));
	P2P_RETURN_TYPE ret = PRT_OK;
	UINT unfinishedSize = 0;
	UINT tmpSize = 0;
	// 1. 读取blockID-1的数据，以便检查blockID块的数据是否正确
	ret = GetBlock(blockID-1, tmpSize, tmpBlockData);
	if(ret != PRT_OK && ret != PRT_BLOCK_NOTFOUND) {// 如果找到或者没找到此块都是正常的，但是其他返回值就不正常了
		assert(0);
		return ret;
	}
	if(ret != PRT_BLOCK_NOTFOUND) {
		// 2. 获取blockID-1块最后一个Sample，在blockID中的剩余长度
		if(!GetUnfinishedSampleSize(tmpBlockData, unfinishedSize)) {
			DelBlock(blockID-1); // 删除错误的blockID-1块
		}
		else if(unfinishedSize != UINT_MAX) {
			// 3. 根据剩余长度检查blockID块数据是否正确
			UINT sampleOffset = *((UINT*)data + 1);
			// TODO: 由于CaptureServer的一个BUG，这里作一个修正，随着所有CaptureServer的更新，这里可以去除。 2005.04.18
			if(sampleOffset == BLOCK_SIZE)
				sampleOffset = UINT_MAX;

			if(sampleOffset != unfinishedSize+sizeof(UINT)*2) {
				if(sampleOffset == UINT_MAX && unfinishedSize > BLOCK_SIZE) {
					; // blockID块被完全跨越，数据应该是正确的
				}
				else if(!IsFirstBlockOfNewProgram(data)) {
					assert(0);
					return PRT_BAD_BLOCK; // 不是新节目，blockID块数据错误，直接错误
				}
			}
		}

		// 4. 读取blockID+1块
		ret = GetBlock(blockID+1, tmpSize, tmpBlockData);
		if(ret != PRT_OK && ret != PRT_BLOCK_NOTFOUND)// 如果找到或者没找到此块都是正常的，但是其他返回值就不正常了
			return ret;
		if(ret != PRT_BLOCK_NOTFOUND) {
			// 5. 获取blockID块最后一个Sample，在blockID+1中的剩余长度
			if(!GetUnfinishedSampleSize(data, unfinishedSize))
				return PRT_OK; // blockID块数据错误，直接返回
			else if(unfinishedSize != UINT_MAX) {
				// 6. 根据剩余长度检查blockID+1块数据是否正确
				UINT sampleOffset = *((UINT*)tmpBlockData + 1);
				// TODO: 由于CaptureServer的一个BUG，这里作一个修正，随着所有CaptureServer的更新，这里可以去除。 2005.04.18
				if(sampleOffset == BLOCK_SIZE)
					sampleOffset = UINT_MAX;

				if(sampleOffset != unfinishedSize+sizeof(UINT)*2) {
					if(sampleOffset == UINT_MAX && unfinishedSize > BLOCK_SIZE) {
						; // blockID+1块被完全跨越，数据应该是正确的
					}
					else if(!IsFirstBlockOfNewProgram(tmpBlockData)) {
						assert(0);
						DelBlock(blockID+1); // 删除错误的blockID+1块, 没有记录此块来自哪个NP
					}
				}
			}
		}
	}


	// 数据正确，存储此Block
	return BaseResource::PutBlock(blockID, blockSize, data);
}

// 获取此Block末尾一个Sample被截断后，在下一个Block所残留的长度，返回值表示此Block是否数据正确
bool LiveResource::GetUnfinishedSampleSize(PBYTE& blockData, UINT& unfinishedSize) {	
	UINT sampleOffset = *((UINT*)blockData + 1);
	if(sampleOffset == UINT_MAX) {
		unfinishedSize = UINT_MAX;
		return true; // 没法检查，因为sample跨越了整个block，放过这种情况吧
	}
	else if(sampleOffset < sizeof(UINT)*2 || sampleOffset > BLOCK_SIZE) {
		assert(0);
		return false; // 数据错误
	}

	SampleHeader* header = (SampleHeader*)(blockData+sampleOffset);
	while(1) {
		if(sampleOffset+sizeof(UINT) > BLOCK_SIZE) {
			// SampleHeader中的Size尚未读全，没法确定残留的长度，放过这种情况
			unfinishedSize = UINT_MAX;
			return true;
		}
		else if(sampleOffset + header->size >= BLOCK_SIZE) {
			unfinishedSize = sampleOffset + header->size - BLOCK_SIZE; // 这就是残留的长度
			if(unfinishedSize > 1000000)
				assert(0);
			return true;
		}
		sampleOffset += header->size;
		header = (SampleHeader*)(blockData+sampleOffset);
	}
	assert(0);
	return false;
}

// 检查此Block是否新节目的第一块
bool LiveResource::IsFirstBlockOfNewProgram(PBYTE& blockData) {
	UINT sampleOffset = *((UINT*)blockData + 1);
	if(sampleOffset == sizeof(UINT)*2) {
		SampleHeader* header = (SampleHeader*)(blockData+sizeof(UINT)*2);
		// 新节目的第一个Sample格式特殊，具体值如下判断
		if(header->length == 0xffffffff && header->start == 0xffffffffffffffff && header->size == sizeof(SampleHeader))
			return true;
	}
	return false;
}

UINT16 LiveResource::GetBufferCount() {
	if(totalBufferCount >= 0xffff)
		return 0xffff;
	return static_cast<UINT16>(totalBufferCount);
}

UINT16 LiveResource::GetBufferTime() {
	if(totalBufferTime/1000 >= 0xffff)
		return 0xffff;
	return static_cast<UINT16>(totalBufferTime/1000);
}

UINT16 LiveResource::GetBufferingTime() {
	// 如果正在缓冲的过程中，那么把到目前为止的时间也要计算上
	DWORD lastGetSampleFailedTime = A_lastGetSampleFailedTime;
	if(lastGetSampleFailedTime == 0)
		lastGetSampleFailedTime = V_lastGetSampleFailedTime;
	if(lastGetSampleFailedTime != 0) {
		UINT bufTime = timeGetTime()-lastGetSampleFailedTime;
		if(bufTime/1000 >= 0xffff)
			return 0xffff;
		return static_cast<UINT16>(bufTime/1000);
	}
	return 0;
}

void LiveResource::SetSPUpdate(const SPUpdate& update, BYTE sum) {
#ifdef _DEBUG
	// TODO: 测试使用，正式版本一定要去掉
	//spUpdate.minBlockID = spUpdate.maxBlockID = 0;
#endif
	// 是否尚未收到过SPUpdate
	bool bNoFirstSPUpdate = (spUpdate.minBlockID == 0 && spUpdate.maxBlockID == 0);
	spUpdate = update;
	spUpdateSum = sum;

	// 是否需要重新设置播放位置
	bool bNeedResetPlayingBlock = bNoFirstSPUpdate;

	// 如果播放位置比SP拥有的最小Block还小，就自动调整播放位置
	if(GetPlayingBlock() < spUpdate.minBlockID)
		bNeedResetPlayingBlock = true;

	// 从TIME_BEGIN_PLAY秒钟前的位置开始下载播放
	if(bNeedResetPlayingBlock) {
		const float SECPERBLOCK = bitRate*1024/BLOCK_SIZE;	// 每块包含的大致时间长度
		const UINT BLOCK_BEFORE = static_cast<UINT>(TIME_BEGIN_PLAY*SECPERBLOCK);	// 提前的块数
		if(spUpdate.maxBlockID >= spUpdate.minBlockID+BLOCK_BEFORE)
			SetPlayingBlock(spUpdate.maxBlockID-BLOCK_BEFORE);
		else
			SetPlayingBlock(spUpdate.minBlockID);
	}
}

BOOL LiveResource::SeekToTime(LONGLONG& targetTime) {
	comm->logFile.StatusOut("Begin Seeking......to %ds %dms", targetTime/10000000, targetTime/10000);
	if(!bInited)
		return FALSE;

	if(targetTime == 0) // reset to begin. do nothing.
		return TRUE;

	if((spUpdate.maxKeySample > spUpdate.minKeySample) && 
		(targetTime >= spUpdate.minKeySample) && (targetTime <= spUpdate.maxKeySample) && 
		spUpdate.maxBlockID > spUpdate.minBlockID) 
	{
		double temp = ((double)(targetTime-spUpdate.minKeySample))/(spUpdate.maxKeySample-spUpdate.minKeySample);
		UINT targetBlockID = static_cast<UINT>(temp*(spUpdate.maxBlockID-spUpdate.minBlockID))+spUpdate.minBlockID;

		SetPlayingBlock(targetBlockID);
	}
	else {
		comm->logFile.StatusOut("Bad seeking position!");
		return FALSE;
	}
	return TRUE;
}

// 添加一个区间对应的媒体类型
void LiveResource::AddMediaInterval(const MediaInterval& mi) {
	comm->logFile.StatusOut("Got MediaType of [%d, %d).", mi.start, mi.start+mi.size);
    // 是否发生了直播改变编码格式
    bool bLiveChangeMediaData = false;
    if(!mediaArray.AddInterval(mi, &bLiveChangeMediaData)) {
		assert(0);
        return;
    }
    if(bLiveChangeMediaData) {
        // 清除NP连接发送媒体类型区间的记录
        comm->p2pMgr.ClearSentMediaArray();

        // 记录不再向上层提交的媒体类型
        MediaInterval temp;
        if(A_sentArray.FindBlock(A_currBlockID)) {
            if(mediaArray.GetInterval(A_currBlockID, temp)) {
                A_sentArray.Clear();
                A_sentArray.AddInterval(temp);
            }
        }
        if(V_sentArray.FindBlock(V_currBlockID)) {
            if(mediaArray.GetInterval(V_currBlockID, temp)) {
                V_sentArray.Clear();
                V_sentArray.AddInterval(temp);
            }
        }
    }
}

// 获取一个Block所在区间的媒体类型
bool LiveResource::GetMediaInterval(const UINT blockID, MediaInterval& mi) {
	return mediaArray.GetInterval(blockID, mi);
}

BOOL LiveResource::SetPlayingBlock(UINT blockID) {
	if(blockID == UINT_MAX)
		return FALSE;

	CriticalSection::Owner lock(dataLocker);
	// 设置开始播放的位置
	A_currBlockID = V_currBlockID = blockID;
	A_leftDataInCurrBlock = V_leftDataInCurrBlock = 0;

	comm->logFile.StatusOut("SetPlayingBlock %d. spupdate.maxblockID %d", blockID, spUpdate.maxBlockID);
	return TRUE;
}

// 对于直播, 取得视频音频中比较大(小)的那个BlockID, 当然,如果是单音频或者单视频的,就取得相应的BlockID
UINT LiveResource::GetPlayingBlock(bool max) {
	if(V_currBlockID == UINT_MAX)
		return A_currBlockID;
	if(A_currBlockID == UINT_MAX)
		return V_currBlockID;

	if(max)
		return max(V_currBlockID, A_currBlockID);
	return min(V_currBlockID, A_currBlockID);
};

int LiveResource::GetBufferPercent() {
	// 要保证blocks4Play不等于0，否则此处一定返回100
	if(blocks4Play == 0)
		return 100;
	UINT nextBlock = GetPlayingBlock();
	return min(100, allIntervals.GetCountInInterval(nextBlock, blocks4Play)*100/blocks4Play);
}

bool LiveResource::EnlargeBuffer() {
	totalBufferCount++;

	DWORD lastGetSampleFailedTime = A_lastGetSampleFailedTime;
	if(lastGetSampleFailedTime == 0)
		lastGetSampleFailedTime = V_lastGetSampleFailedTime;
	if(lastGetSampleFailedTime != 0)
		totalBufferTime += (timeGetTime() - lastGetSampleFailedTime);

	if(blocks4Play >= static_cast<UINT>(180*bitRate*1024/BLOCK_SIZE)) {
		comm->logFile.StatusOut("Delay more than 3 minutes, do not add.");
		return false;
	}
	comm->logFile.StatusOut("live Pause! Add 50%% time for buffer.");
	blocks4Play = static_cast<UINT>(blocks4Play*1.5f);
	return true;
}

void LiveResource::PrintStatus() {
	comm->logFile.StatusOut("CurrBlock: A%d/V%d, Continued: %d/%d. BlockRange: %d/%d, SPUpdate: %d/%d, \n\tBuffer Percent: %d%%. BufferCount: %d, BufferTime: %ds, curr Buffer: %ds.", 
		A_currBlockID, V_currBlockID, 
		allIntervals.GetContinousCount(GetPlayingBlock()), 
		allIntervals.GetCountInInterval(GetPlayingBlock(), UINT_MAX), 
		GetMinBlockID(), GetMaxBlockID(), 
		spUpdate.minBlockID, spUpdate.maxBlockID, 
		GetBufferPercent(), GetBufferCount(), 
		GetBufferTime(), GetBufferingTime());
}

void LiveResource::SetDefaultCP() {
	// 直播的默认CP地址就是和SP相同的IP
	if(spIPList) {
		PeerInfoWithAddr nAddr;
		nAddr.outerIP.sin_addr.s_addr = spIPList[0].sin_addr.s_addr;
		nAddr.subnetIP.sin_addr.s_addr = 0xffffffff;
		nAddr.subnetIP.sin_port = htons(CP4NP_PORT);
		nAddr.isCachePeer = true;
		nAddr.layer = 0;

		comm->p2pMgr.AddPeerInfo(nAddr);

		comm->logFile.StatusOut("Choose Default CP %s.", comm->p2pMgr.FormatIPAddress(nAddr));
	}
};

// 获取节目名字
string LiveResource::GetProgramName() {
	MediaInterval mInterval;
	if(!mediaArray.GetInterval(GetPlayingBlock(), mInterval))
		return "";
	if(!mInterval.pname || !mInterval.pnamesize)
		return "";
	string temp = mInterval.pname;
	return temp;
}

// 获取节目时间长度
UINT32 LiveResource::GetProgramTimeInSeconds() {
	MediaInterval mInterval;
	if(!mediaArray.GetInterval(GetPlayingBlock(), mInterval))
		return 0;
	return mInterval.progtime;
}

// 获取频道名字
string LiveResource::GetChannelName() {
	MediaInterval mInterval;
	if(!mediaArray.GetInterval(GetPlayingBlock(), mInterval))
		return "";
	if(!mInterval.cname || !mInterval.cnamesize)
		return "";
	string temp = mInterval.cname;
	return temp;
}
}