acmtransformer.cxx

rtp在linux下的实现

// ACMTransformer.cpp: implementation of the ACMTransformer class.
//
//////////////////////////////////////////////////////////////////////

#include <windows.h>
#include <algorithm>

#include "ACMTransformer.h"
#include "AudioSampleManager.h"
#include "AudioSample.h"

using namespace std;

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

ACMTransformer::ACMTransformer()
{
	SetSourceFormat(WaveFormat::GetWaveFormat(WaveFormat_PCM_16_8_1));
	SetDestinationFormat(WaveFormat::GetWaveFormat(WaveFormat_PCM_16_8_1));
	InitializeCriticalSection(&outputQueueMutex);
	InitializeCriticalSection(&freeBuffersMutex);
	InitializeCriticalSection(&stateMutex);
	freeBuffersEvent	= CreateEvent(NULL, FALSE, FALSE, NULL);
	outputEnqueueEvent	= CreateEvent(NULL, FALSE, FALSE, NULL);
	outputDequeueEvent	= CreateEvent(NULL, FALSE, FALSE, NULL);
	SetOutputDuration(20);
	char subFacilityName[100];
	sprintf(subFacilityName, "ACMTransformer:%x", this);
	tracer.SetSubFacilityName(subFacilityName);
	SetRunning(false);
	SetTraceLevel();
}

ACMTransformer::~ACMTransformer()
{
	DeleteCriticalSection(&outputQueueMutex);
	DeleteCriticalSection(&freeBuffersMutex);
	DeleteCriticalSection(&stateMutex);
	CloseHandle(freeBuffersEvent);
	CloseHandle(outputEnqueueEvent);
	CloseHandle(outputDequeueEvent);
}

int
ACMTransformer::SetTraceLevel()
{
	long SystemMask = 0;
	if ((SystemMask = GetRegKeyLong(HKEY_CURRENT_USER, "Software\\Cisco Systems\\MTC\\Tracing", "AllComponents", 0x0)) == 0)
	{
		SystemMask = GetRegKeyLong(HKEY_CURRENT_USER, "Software\\Cisco Systems\\MTC\\Tracing", "ACMTransformer", 0x100000);
	}
	tracer.SetSystemMask(SystemMask);
	return 0;
}

int
ACMTransformer::SetSourceFormat(WAVEFORMATEX &wfxSource)
{
	sourceFormat = wfxSource;
	return 0;
}

int
ACMTransformer::SetDestinationFormat(WAVEFORMATEX &wfxDest)
{
	destFormat = wfxDest;
	return 0;
}

int
ACMTransformer::StartTransform()
{
	int result(0);
	result = acmStreamOpen(&hACMStream, NULL, &sourceFormat, &destFormat, NULL, 0, 0, ACM_STREAMOPENF_NONREALTIME);
	if (result == 0)
	{
		for (int i=0; i<ACMTRANSFORMER_NUM_STREAMHEADER; i++)
		{
			memset(&acmStreamHeader[i], 0, sizeof(acmStreamHeader[i]));
			acmStreamHeader[i].cbStruct			= sizeof(acmStreamHeader[i]);
			acmStreamHeader[i].fdwStatus		= 0;
			acmStreamHeader[i].pbSrc			= sourceBuffer[i];
			acmStreamHeader[i].cbSrcLength		= ACMTRANSFORMER_MAX_SOURCEBUFFERSIZE;
			acmStreamHeader[i].pbDst			= destBuffer[i];
			acmStreamHeader[i].cbDstLength		= ACMTRANSFORMER_MAX_DESTBUFFERSIZE;
			result = acmStreamPrepareHeader(hACMStream, &acmStreamHeader[i], 0);
			acmStreamHeader[i].cbSrcLength		= 0;	// to show that no unconverted data
			acmStreamHeader[i].cbSrcLengthUsed	= 0;
			if (result == 0)
			{
				ReleaseBuffer(&acmStreamHeader[i]);	// adds it to the free buffer
				SetRunning(true);
			}
		}
	}
	return result;
}

int
ACMTransformer::StopTransform()
{
	// if there is no input pin, then wait for buffered data to clear
	int numSources	= audioSources.size();
	int numDest		= audioSinks.size();
	if (numSources == 0 && numDest > 0)
	{
		AudioSample *audioSample;
		AudioSampleManager::GetInstance()->GetAudioSample(&audioSample);
		if (audioSample)
		{
			audioSample->SetFormat(destFormat);
			int minFrameSize = (audioSample->MinFrameSize() * outputDuration * 1000) / audioSample->MinFrameDuration();
			while (true)
			{
				EnterCriticalSection(&outputQueueMutex);
				if (outputQueue.size() < minFrameSize)
				{
					LeaveCriticalSection(&outputQueueMutex);
					break;
				}
				LeaveCriticalSection(&outputQueueMutex);
				WaitForSingleObject(outputDequeueEvent, INFINITE);
			}
			audioSample->Release();
		}
	}

	int result(0);
	SetRunning(false);
	EnterCriticalSection(&freeBuffersMutex);
	freeBuffers.clear();
	SetEvent(freeBuffersEvent);
	LeaveCriticalSection(&freeBuffersMutex);

	EnterCriticalSection(&outputQueueMutex);
	SetEvent(outputEnqueueEvent);
	SetEvent(outputDequeueEvent);
	LeaveCriticalSection(&outputQueueMutex);
	
	for (int i=0; i<ACMTRANSFORMER_NUM_STREAMHEADER; i++)
	{
		acmStreamHeader[i].cbSrcLength		= ACMTRANSFORMER_MAX_SOURCEBUFFERSIZE;
		acmStreamHeader[i].cbDstLength		= ACMTRANSFORMER_MAX_DESTBUFFERSIZE;
		result = acmStreamUnprepareHeader(hACMStream, &acmStreamHeader[i], 0);
	}
	result = acmStreamClose(hACMStream, 0);
	return result;
}

int
ACMTransformer::TransformAudioSamples(std::vector<std::pair<AudioSample *, AudioSource *> > &data, AudioSample **ppAudioSample)
{
	return 0;
}

bool
ACMTransformer::IsRunning()
{
	EnterCriticalSection(&stateMutex);
	bool result = bRunning;
	LeaveCriticalSection(&stateMutex);
	return result;
}

void
ACMTransformer::SetRunning(bool running)
{
	EnterCriticalSection(&stateMutex);
	bRunning = running;
	LeaveCriticalSection(&stateMutex);
}

ACMSTREAMHEADER *
ACMTransformer::GetBuffer()
{
	ACMSTREAMHEADER *acmHdr = NULL;
	while(IsRunning())
	{
		EnterCriticalSection(&freeBuffersMutex);
		if (freeBuffers.size() > 0) 
		{
			acmHdr = freeBuffers.front();
			freeBuffers.pop_front();
			LeaveCriticalSection(&freeBuffersMutex);
			break;
		}
		else 
		{
			LeaveCriticalSection(&freeBuffersMutex);
			WaitForSingleObject(freeBuffersEvent, INFINITE);
		}
	}
	return acmHdr;
}

void
ACMTransformer::ReleaseBuffer(ACMSTREAMHEADER *acmHdr)
{
	EnterCriticalSection(&freeBuffersMutex);
	freeBuffers.push_back(acmHdr);
	SetEvent(freeBuffersEvent);
	LeaveCriticalSection(&freeBuffersMutex);
}

int
ACMTransformer::RenderAudioSamples(std::vector<std::pair<AudioSample *, AudioSource *> > &data)
{
	tracer.tracef(DET, "RenderAudioSamples\n");
	int result(0);
	AudioSample *inSample = data[0].first;
	if (inSample)
	{
		if (inSample->DataSize() > 0)
		{
			ACMSTREAMHEADER *acmHdr = GetBuffer();
			if (acmHdr)
			{
				int extraBytes = acmHdr->cbSrcLength - acmHdr->cbSrcLengthUsed;
				int totalBytes = inSample->DataSize() + extraBytes;
				if ( totalBytes > 0 && totalBytes <= ACMTRANSFORMER_MAX_SOURCEBUFFERSIZE)
				{
					acmHdr->cbSrcLength = totalBytes;
					memcpy(acmHdr->pbSrc, acmHdr->pbSrc + acmHdr->cbSrcLengthUsed, extraBytes);
					memcpy(acmHdr->pbSrc + extraBytes, inSample->Data(), inSample->DataSize());
					result = acmStreamConvert(hACMStream, acmHdr, ACM_STREAMCONVERTF_BLOCKALIGN);
					if (result != 0)
					{
						result = -10;
						Sleep(outputDuration);
					}
					else if (acmHdr->cbDstLengthUsed > 0)
					{
						AddConvertedData(acmHdr->pbDst, acmHdr->cbDstLengthUsed);
					}
					ReleaseBuffer(acmHdr);
				}
				else
				{
					Sleep(outputDuration);
					result = -20;
				}
			}
			else
			{
				Sleep(outputDuration);
				result = -30;
			}
		}
		else
		{
			Sleep(inSample->GetSilenceDuration());
			result = -40;
		}
	}
	else
	{
		Sleep(outputDuration);
		result = -50;
	}
	tracer.tracef(DET, "~RenderAudioSamples : %d\n", result);
	return result;
}

int
ACMTransformer::GenerateData(AudioSample **ppAudioSample)
{
	tracer.tracef(DET, "GenerateData\n");
	int result(0);
	*ppAudioSample = NULL;
	AudioSample *audioSample;
	(AudioSampleManager::GetInstance())->GetAudioSample(&audioSample);
	if (audioSample)
	{
		*ppAudioSample = audioSample;
		audioSample->SetFormat(destFormat);
		audioSample->SetDataSize(0);
		audioSample->SetSilenceDuration(outputDuration);
		int numBytesToWrite = (audioSample->MinFrameSize() * outputDuration * 1000) / audioSample->MinFrameDuration();

		if (IsRunning())
		{
			EnterCriticalSection(&outputQueueMutex);
			if (outputQueue.size() < numBytesToWrite)
			{
				LeaveCriticalSection(&outputQueueMutex);
				result = -10;
			}
			else
			{
				int numBytesToCopy = min(numBytesToWrite, audioSample->BufferSize());
				for (int i=0; i<numBytesToCopy; i++)
				{
					unsigned char c = outputQueue.front();
					outputQueue.pop();
					(audioSample->Data())[i] = c;
				}
				SetEvent(outputDequeueEvent);
				LeaveCriticalSection(&outputQueueMutex);
				audioSample->SetDataSize(numBytesToCopy);
			}
		}
		else
		{
			EnterCriticalSection(&outputQueueMutex);
			SetEvent(outputDequeueEvent);
			LeaveCriticalSection(&outputQueueMutex);
			result = -20;
		}
	}
	else
	{
		result = -30;
	}
	tracer.tracef(DET, "~GenerateData : %d\n", result);
	return result;
}

int
ACMTransformer::AddConvertedData(unsigned char *dataIn, int numBytes)
{
	while (true)
	{
		if (IsRunning())
		{
			EnterCriticalSection(&outputQueueMutex);
			int queueSpace = ACMTRANSFORMER_OUTPUTBUFFERSIZE - outputQueue.size();
			if (queueSpace >= numBytes)
			{
				for (int i=0; i<numBytes; i++)
				{
					outputQueue.push(dataIn[i]);
				}
				SetEvent(outputEnqueueEvent);
				LeaveCriticalSection(&outputQueueMutex);
				break;
			}
			else
			{
				LeaveCriticalSection(&outputQueueMutex);
				WaitForSingleObject(outputDequeueEvent, INFINITE);
			}
		}
		else
		{
			EnterCriticalSection(&outputQueueMutex);
			SetEvent(outputEnqueueEvent);
			LeaveCriticalSection(&outputQueueMutex);
			break;
		}
	}
	return 0;
}

int
ACMTransformer::SetOutputDuration(int millisecs)
{
	outputDuration = millisecs;
	return 0;
}

void CALLBACK 
ACMTransformer::ACMCallback(HACMSTREAM has, UINT uMsg, DWORD dwInstance, LPARAM lParam1, LPARAM lParam2)
{
	ACMTransformer	*acmTransformer	= (ACMTransformer  *)dwInstance;
	ACMSTREAMHEADER	*acmHdr			= (ACMSTREAMHEADER *)lParam1;
	switch(uMsg)
	{
	case MM_ACM_OPEN:
		break;
	case MM_ACM_CLOSE:
		break;
	case MM_ACM_DONE:
		if (acmHdr->cbDstLengthUsed > 0)
		{
			acmTransformer->AddConvertedData(acmHdr->pbDst, acmHdr->cbDstLengthUsed);
		}
		acmTransformer->ReleaseBuffer(acmHdr);
		break;
	default:
		break;
	}
}