buffermgr.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
*
*/
// BufferMgr.cpp: implementation of the BufferMgr class.
//
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"
#include "BufferMgr.h"
#include "CaptureServer.h"
#include "LogMgr.h"

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////

BufferMgr::BufferMgr(CaptureServer* c) : cs(c), blockCount(0), newBlock(0), newBlockSize(0), 
		blockArray(0), blockSize(0), maxBlockNum(0), maxBlockID(0), 
		firstKeySample(0), sampleBuffer(0), sampleBufferSize(0) {
	sampleStartTime = 0;
    bSwitchMedia = false;

	// TEST: 检测时光倒退的问题
	lastSampleStart = 0;
	lastRecvSampleTime = 0;
	startseconds = 0;
	
	RemoveOldTmpFile();

	bShouldSave = FALSE;
	bShouldConnect = FALSE;

	char buf[MAX_PATH+1];
	DWORD res = GetTempPath(MAX_PATH, buf);
	bufferPath = buf;

	UINT uni = GetTickCount();
	GetTempFileName(bufferPath.data(), "#CD", uni, buf);
	bufferPath = buf;
	if(!ExCreateFile(hBufferFile, buf, CREATE_ALWAYS))
	  return;
	ExSetFileSize(hBufferFile, 0);

	newBlock = new BYTE[BLOCK_SIZE];
	maxBlockNum = BUFFER_SPACE / BLOCK_SIZE;
	blockArray = new UINT[maxBlockNum];
	for(UINT i = 0; i < maxBlockNum; i++)
		blockArray[i] = UINT_MAX;
	blockSize = new UINT[maxBlockNum];
}

BufferMgr::~BufferMgr() {
	CAutoLock lock(&bufferfile_cs);
	StopSave();
	CloseHandle(hBufferFile);
	DeleteFile(bufferPath.data());

	SAFE_ARRAYDELETE(blockArray);
	SAFE_ARRAYDELETE(blockSize);
	SAFE_ARRAYDELETE(newBlock);
	SAFE_ARRAYDELETE(sampleBuffer);
}

void BufferMgr::RemoveOldTmpFile() {
	WIN32_FIND_DATA fileData;

	char buf[MAX_PATH+1];
	DWORD res = GetTempPath(MAX_PATH, buf);

	string match = buf;
	match.append("#CD*.tmp");
	HANDLE hFind = FindFirstFile(match.data(), &fileData);
	if(hFind == INVALID_HANDLE_VALUE)
		return;

	while(1) {
		string path = buf;
		path.append(fileData.cFileName);
		DeleteFile(path.data());

		if(!FindNextFile(hFind, &fileData)) {
			if(GetLastError() == ERROR_NO_MORE_FILES)
				break;
			else
				return; 
		}
	}
	FindClose(hFind);
}

void BufferMgr::StartSave() {
	bShouldSave = TRUE;
}

void BufferMgr::StopSave() {
	bShouldSave = FALSE;
	if(blockCount > 0) {
		// 将已经存储的数据清除
		blockCount = 0;
		for(UINT i = 0; i < maxBlockNum; i++)
			blockArray[i] = UINT_MAX;
	}
	maxBlockID = 0;
	firstKeySample = FALSE;
}

/*
 *	block content
 *  |offset of first keysample(int32)|offset of first sample(int32)|list of samples|last uncomplete sample|
 *  |sample data| = |header(SampleHeader)|data(...)|
 */
BOOL BufferMgr::SaveSample(UINT dataOff, const UINT allSize) {
	// at the start of new block, write the offset of next sample
	if(newBlockSize == 0) {
		// 在开头的4个字节写入first keysample offset
		*(UINT*)newBlock = 0;	// 默认值是0
		// 在此后的4个字节写入first sample offset
		*((UINT*)newBlock+1) = sizeof(int)*2;		// 如果开始保存新的Sample，则first sample offset = 8
		if(dataOff > 0)
			*((UINT*)newBlock+1) += allSize-dataOff;// 如果保存前一个sample剩下的部分，则要加上其剩下的长度
		if(*((UINT*)newBlock+1) >= BLOCK_SIZE)		// 如果此sample剩下的部分长度超过当前Block，则用UINT_MAX表示
			*((UINT*)newBlock+1) = UINT_MAX;

		newBlockSize += sizeof(int)*2;
	}

	if(	*(UINT*)(newBlock) == 0 &&		// 当前Block尚未记录FirstKeySampleOffset
		dataOff == 0 &&					// 一个新的Sample
		firstKeySample &&				// 这个Sample是KeySample
		newBlockSize+sizeof(SampleHeader) < BLOCK_SIZE) // SampleHeader刚好保存在当前Block中
	{
		// 在开头的4个字节记录FirstKeySampleOffset, sizeof(UINT)*2是start在SampleHeader中的位置
		*(UINT*)(newBlock) = newBlockSize+sizeof(UINT)*2;
	}

	// 比较剩余数据与剩余空间
	if(allSize-dataOff >= BLOCK_SIZE-newBlockSize) {
		// 剩余数据超过剩余空间，则填满并保存当前Block，并继续存储剩下的数据
		memcpy(newBlock+newBlockSize, sampleBuffer+dataOff, BLOCK_SIZE-newBlockSize);
		dataOff += BLOCK_SIZE-newBlockSize;

		newBlockSize = 0;	// 开始新的Block
		// 保存旧的Block
		if(!SaveNewBlock(newBlock, BLOCK_SIZE))
			return FALSE;

		// 继续保存剩余的数据
		if(allSize-dataOff > 0)
			return SaveSample(dataOff, allSize);
	}
	else {
		// 剩余数据小于剩余空间，复制并等待下一个Sample
		memcpy(newBlock+newBlockSize, sampleBuffer+dataOff, allSize-dataOff);
		newBlockSize += allSize-dataOff;
	}
	return TRUE;
}

BOOL BufferMgr::PutSample(const SampleHeader& header, BYTE* pData, LogMgr* log) {
	CAutoLock lock(&bufferfile_cs);

    if(bSwitchMedia) {
        bSwitchMedia = false;
        // 因为切换编码，首先填入“新节目标志”Sample
        SampleHeader header;
        memset(&header, 0, sizeof(SampleHeader));
        header.size = sizeof(SampleHeader);
        header.length = 0xffffffff;
        header.start = 0xffffffffffffffff;
        header.bSyncPoint = 1;
        // copy sample header into sample buffer
        if(header.size > sampleBufferSize) {
            SAFE_ARRAYDELETE(sampleBuffer);
            sampleBuffer = new BYTE[header.size];
            sampleBufferSize = header.size;
        }
        memcpy(sampleBuffer, &header, sizeof(SampleHeader));
        if(!SaveSample(0, header.size))
            return FALSE;
    }

	// 记录接收到Sample的时间
	lastRecvSampleTime = time(NULL);

	time_t nowtime;
	time(&nowtime);
	if(startseconds == 0)
		startseconds = nowtime;
	if((nowtime - startseconds < TIME4WAIT2STORE))
		return TRUE;

	bShouldConnect = TRUE;

	if(!bShouldSave)
		return TRUE;

	if(!pData || header.size > 1024*1024)
		return FALSE;

	char tmpStr[96];
	_i64toa(header.start, tmpStr, 10);
	_i64toa(header.length+header.start, tmpStr+32, 10);
    //log->StatusOut("recv %s sample start: %s end: %s, put at %d", header.bAudioSample?"audio":"video", 
    //    tmpStr, tmpStr+32, newBlockSize);

	if(sampleStartTime == 0) {
		time_t temp;
		time(&temp);
		sampleStartTime = (LONGLONG)temp*10000000;
		DbgLog((LOG_TRACE, 5, TEXT("此刻 %s！"), ctime(&temp)));

		// 校正时间，因为开始录制的时间并不是此刻,而是打在sample中的时间。
		sampleStartTime -= header.start;
		temp = static_cast<time_t>(sampleStartTime/10000000);
		DbgLog((LOG_TRACE, 5, TEXT("实际开始采集的时间 %s！"), ctime(&temp)));
	}

	if(header.bAudioSample) {
		if(header.start+header.length <= lastSampleStart) {
			char tmpStr[96];
			_i64toa(lastSampleStart, tmpStr, 10);
			_i64toa(header.start, tmpStr+32, 10);
			_i64toa(header.start+header.length, tmpStr+64, 10);
			DbgLog((LOG_TRACE, 5, TEXT("哇！时光倒流，从%s变到了%s->%s！"), tmpStr, tmpStr+32, tmpStr+64));
		}
		lastSampleStart = header.start;
	}


	// record time of the frist keysample of current block
	if(firstKeySample == FALSE && header.bSyncPoint) 
		firstKeySample = TRUE;

	if(header.size > sampleBufferSize) {
		SAFE_ARRAYDELETE(sampleBuffer);
		sampleBuffer = new BYTE[header.size];
		sampleBufferSize = header.size;
	}

	// copy sample header into sample buffer
	memcpy(sampleBuffer, &header, sizeof(SampleHeader));
	if(!cs->GetIsAudioOnly()) {
		((SampleHeader*)sampleBuffer)->start += sampleStartTime;
	}

	// copy sample data into sample buffer;
	memcpy(sampleBuffer+sizeof(SampleHeader), pData, header.size-sizeof(SampleHeader));

	BOOL ret = SaveSample(0, header.size);		
	return ret;
}

BOOL BufferMgr::SaveNewBlock(BYTE* buf, UINT size) {
	if(!buf || size > BLOCK_SIZE)
		return FALSE;

	BOOL ret = TRUE;
	UINT minBlock = UINT_MAX;
	UINT minIndex = 0;
	for(UINT i = 0; i < maxBlockNum; i++) {
		if(blockArray[i] == UINT_MAX) // empty block
			break;
		if(minBlock > blockArray[i]) {
			minBlock = blockArray[i];
			minIndex = i;
		}
	}
	if(i == maxBlockNum) {// no empty block found，replace the min block
		i = minIndex;
	}
	
	if(!ExSetFilePointer(hBufferFile, i*BLOCK_SIZE))
		ret = FALSE;
	else if(!ExWriteFile(hBufferFile, buf, size))
		ret = FALSE;

	if(ret) {
		blockArray[i] = blockCount; 
		blockSize[i] = size;
		blockCount++; // blockCount-1 is blockID
		// 等待下一个Block的first 可以sample
		firstKeySample = FALSE;

		maxBlockID = max(blockArray[i], maxBlockID);

        // 如果当前块有编码类型，则保存当前块的编码类型
        if(!currMediaData.IsEmpty()) {
            mediaMap.insert(pair<UINT, MediaData>(blockArray[i], currMediaData));

            memset(&currMediaData.audioType, 0, sizeof(currMediaData.audioType));
            memset(&currMediaData.videoType, 0, sizeof(currMediaData.videoType));
            delete [] currMediaData.audioData;
            currMediaData.audioData = NULL;
            delete [] currMediaData.videoData;
            currMediaData.videoData = NULL;
        }
	}

	return ret;
}

BOOL BufferMgr::GetBlock(UINT blockID, BYTE* buf, UINT& size) {
	CAutoLock lock(&bufferfile_cs);
	if(!buf || blockID > GetPlayingBlock())
		return FALSE;
	BOOL ret = TRUE;

	for(UINT i = 0; i < maxBlockNum; i++) {
		if(blockArray[i] == blockID && blockArray[i] != UINT_MAX)
			break;
	}
	if(i == maxBlockNum) // not found
		ret = FALSE;

	if(ret) {
		size = blockSize[i];
		if(!ExSetFilePointer(hBufferFile, i*BLOCK_SIZE))
			ret = FALSE;
		else if(!ExReadFile(hBufferFile, buf, size))
			ret = FALSE;
	}
	return ret;
}

// 获取一个块的编码类型
BOOL BufferMgr::GetMediaData(UINT blockID, MediaData& data) {
    map<UINT, MediaData>::const_iterator cit = mediaMap.find(blockID);
    if(cit == mediaMap.end())
        return FALSE;
    if(cit->second.IsEmpty()) {
        ASSERT(0);
        return FALSE;
    }
    data = cit->second;
    return TRUE;
}

// 标志切换编码的Block
BOOL BufferMgr::AttachMediaDataToCurrentBlock(const TVMEDIATYPESECTION& tv, const BYTE* data, BOOL bAudio, LogMgr* log) {
	CAutoLock lock(&bufferfile_cs);

    if(bAudio) {
        currMediaData.audioType = tv;
        currMediaData.audioData = new BYTE[currMediaData.audioType.cbFormat];
        memcpy(currMediaData.audioData, data, currMediaData.audioType.cbFormat);
    }
    else {
        currMediaData.videoType = tv;
        currMediaData.videoData = new BYTE[currMediaData.videoType.cbFormat];
        memcpy(currMediaData.videoData, data, currMediaData.videoType.cbFormat);
    }
    // 清空当前block，准备开始保存新的编码方式的block
    newBlockSize = 0;
    bSwitchMedia = true;

    log->StatusOut("attach %s data to block %d.", bAudio?"audio":"video", blockCount);
    return TRUE;
}

BOOL BufferMgr::ExSetFileSize(HANDLE handle, int size) {
	BOOL ret = TRUE;
	if(!ExSetFilePointer(handle, size))
		ret = FALSE;
	else if(0 == SetEndOfFile(handle)) {
		ret = FALSE;
	}
	return ret;
}

int BufferMgr::ExReadFile(HANDLE handle, LPVOID buf, int toBeRead) {
	int readCount = 0;
	DWORD tmpRead;
	while(readCount < toBeRead) {
		BOOL success = ReadFile(
				handle,					// file handler
				(char*)buf+readCount,	// current buffer position
				toBeRead - readCount,	// remaining bytes to be read
				&tmpRead,				// read bytes
				NULL);					// not overlaped
		if(!success)
			return FALSE;
		if(tmpRead == 0) // no more data
			return FALSE;
		readCount += tmpRead;
	}
	return readCount;
}

BOOL BufferMgr::ExWriteFile(HANDLE handle, LPVOID buf, int toBeWrite) {
	int writeCount = 0;
	DWORD tmpWrite;
	while(writeCount < toBeWrite) {
		BOOL success = WriteFile(
				handle,					// file handler
				(char*)buf+writeCount,	// current buffer position
				toBeWrite-writeCount,	// remaining bytes to be read
				&tmpWrite,				// written bytes
				NULL);					// not overlaped
		if(!success) {
			return FALSE;
		}
		writeCount += tmpWrite;
	}
	return writeCount;
}

BOOL BufferMgr::ExSetFilePointer(HANDLE handle, int offset) {
	if(-1 == SetFilePointer(handle, 
							offset, 
							NULL, 
							FILE_BEGIN)) {
		return FALSE;
	}
	return TRUE;
}

BOOL BufferMgr::ExCreateFile(HANDLE& handle,
							LPCTSTR lpFileName,
							DWORD dwDesiredAccess,
							DWORD dwShareMode,
							LPSECURITY_ATTRIBUTES lpSecurityAttributes,
							DWORD dwCreationDisposition,
							DWORD dwFlagsAndAttributes,
							HANDLE hTemplateFile) {
	handle = CreateFile(lpFileName, dwDesiredAccess, dwShareMode, lpSecurityAttributes, 
						dwCreationDisposition, dwFlagsAndAttributes, hTemplateFile);
	if(handle == INVALID_HANDLE_VALUE) {
		return FALSE;
	}
	return TRUE;
}

BOOL BufferMgr::ExCreateFile(HANDLE& handle, LPCTSTR lpFileName, DWORD dwCreationDisposition) {
	handle = CreateFile(lpFileName, 
						GENERIC_WRITE | GENERIC_READ, 
						FILE_SHARE_READ | FILE_SHARE_WRITE, 
						NULL, 
						dwCreationDisposition, 
						FILE_ATTRIBUTE_NORMAL, 
						NULL);
	if(handle == INVALID_HANDLE_VALUE) {
		return FALSE; 
	}
	return TRUE;
}