audiosink.cxx

rtp在linux下的实现

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

#include <algorithm>

#include "AudioSink.h"
#include "AudioSource.h"
#include "AudioSample.h"

using namespace std;

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

AudioSink::AudioSink()
{
	bSinkRunning = false;
	bSinkThreadRunning = false;
#ifdef WIN32
	sinkThreadHandle = NULL;
	sinkThreadID = 0;
	sinkThreadStartEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
	sinkThreadStopEvent  = CreateEvent(NULL, FALSE, FALSE, NULL);
	InitializeCriticalSection(&audioSourcesMutex);
	InitializeCriticalSection(&sinkThreadMutex);
	InitializeCriticalSection(&sinkStateMutex);
#endif
}

AudioSink::~AudioSink()
{
#ifdef WIN32
	CloseHandle(sinkThreadStartEvent);
	CloseHandle(sinkThreadStopEvent);
	DeleteCriticalSection(&audioSourcesMutex);
	DeleteCriticalSection(&sinkThreadMutex);
	DeleteCriticalSection(&sinkStateMutex);
#endif
}

int
AudioSink::SubscribeAudioSource(AudioSource *audioSource)
{
	EnterCriticalSection(&audioSourcesMutex);
	if (find(audioSources.begin(), audioSources.end(), audioSource) == audioSources.end())
	{
		audioSources.push_back(audioSource);
	}
	LeaveCriticalSection(&audioSourcesMutex);
	return 0;
}

int
AudioSink::UnsubscribeAudioSource(AudioSource *audioSource)
{
	EnterCriticalSection(&audioSourcesMutex);
	vector<AudioSource *>::iterator position = NULL;
	position = find(audioSources.begin(), audioSources.end(), audioSource);
	if (position != audioSources.end())
	{
		audioSources.erase(position);
	}
	LeaveCriticalSection(&audioSourcesMutex);
	return 0;
}

int
AudioSink::RenderAudioSamples(std::vector<std::pair<AudioSample *, AudioSource *> > &data)
{
	tracer.tracef(ERR, "RenderAudioSamples : ERROR : pure virtual function call !!!\n");
	return 0;
}

int
AudioSink::TakeNextAudioSample(AudioSample *audioSample, AudioSource *audioSource)
{
	int result;
	audioSample->AddRef();	// we later release it
	inputData.clear();
	inputData.push_back(make_pair(audioSample, audioSource));
	EnterCriticalSection(&audioSourcesMutex);
	vector<AudioSource *>::iterator iter = audioSources.begin();
	for (iter=audioSources.begin(); iter!=audioSources.end(); iter++)
	{
		if (*iter == audioSource)
		{
		tracer.tracef(DET, "TakeNextAudioSample from the same source \n");
			continue;
		}
		AudioSample *pAudioSample;
		tracer.tracef(DET, "TakeNextAudioSample : calling GiveNextAudioSample\n");
		(*iter)->GiveNextAudioSample(&pAudioSample, this);
                if(pAudioSample)
		    inputData.push_back(make_pair(pAudioSample, *iter));
	}
	LeaveCriticalSection(&audioSourcesMutex);
	result = RenderAudioSamples(inputData);
	for (int i=0; i<inputData.size(); i++)
	{
		AudioSample *sample = inputData[i].first;
		sample->Release();
	}
	return result;
}

int
AudioSink::StartSink()
{
	int result(0);
	EnterCriticalSection(&sinkStateMutex);
	if (bSinkThreadRunning || bSinkRunning)
	{
		result = -10;
	}
	else
	{
		bSinkRunning = true;
	}
	LeaveCriticalSection(&sinkStateMutex);
	return result;
}

int
AudioSink::StopSink()
{
	int result(0);
	EnterCriticalSection(&sinkStateMutex);
	if (bSinkThreadRunning)
	{
		result = -10;
	}
	else
	{
		bSinkRunning = false;
	}
	LeaveCriticalSection(&sinkStateMutex);
	return result;
}

int
AudioSink::StartSinkThread()
{
	int result(0);
#ifdef WIN32
	int returnCode(0);
#endif
	tracer.tracef(EE, "StartSinkThread\n");
	EnterCriticalSection(&sinkStateMutex);
	tracer.tracef(ARB, "StartSinkThread : entered sinkStateMutex\n");
	do
	{
		if (bSinkRunning || bSinkThreadRunning)
		{
			tracer.tracef(ERR, "StartSinkThread : e-error : bSinkRunning = %d, bSinkThreadRunning = %d\n", bSinkRunning, bSinkThreadRunning);
			result = -10;
			break;
		}
#ifdef WIN32
		if (sinkThreadHandle != NULL || sinkThreadID != 0)
		{
			tracer.tracef(ERR, "StartSinkThread : e-error : sinkThreadHandle = 0x%x, sinkThreadID = 0x%x\n", sinkThreadHandle, sinkThreadID);
			result = -20;
			break;
		}
		ResetEvent(sinkThreadStartEvent);
		ResetEvent(sinkThreadStopEvent);
#endif
		bSinkThreadShouldStop = true;		// we make it false only if everything else succeeds
#ifdef WIN32
		sinkThreadHandle = CreateThread(NULL, 0, SinkThreadProc, (void *)this, CREATE_SUSPENDED, &sinkThreadID);
		if (sinkThreadHandle == NULL)
		{
			tracer.tracef(ERR, "StartSinkThread : e-CreateThread\n");
			result = -30;
			break;
		}
		returnCode = SinkThreadStarted(sinkThreadHandle, sinkThreadID);
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StartSinkThread : e-SinkThreadStarted : %d 0x%x\n", returnCode, returnCode);
			result = -40;
			break;
		}
		bSinkThreadShouldStop = false;
		returnCode = ResumeThread(sinkThreadHandle);
		if (result == -1)
		{
			tracer.tracef(ERR, "StartSinkThread : e-ResumeThread : %d 0x%x\n", returnCode, returnCode);
			result = -50;
			break;
		}
		WaitForSingleObject(sinkThreadStartEvent, INFINITE);
#else
		bSinkThreadShouldStop = false;
		sinkThreadHandle.spawn(SinkThreadProc, (void *)this);
                //sinkThreadStartEvent.wait(&sinkStateMutex, -1);
		SinkThreadStarted(*sinkThreadHandle.getId(), *sinkThreadHandle.getId());
#endif
	}
	while (false);
	LeaveCriticalSection(&sinkStateMutex);
	tracer.tracef(ARB, "StartSinkThread : left  sinkStateMutex\n");
	tracer.tracef(EE, "~StartSinkThread : returning %d\n", result);
	return result;
}

int
AudioSink::StopSinkThread()
{
	int result(0);
#ifdef WIN32
	int returnCode(0);
#endif
	bool waitForThread = true;
	tracer.tracef(EE, "StopSinkThread\n");
	EnterCriticalSection(&sinkStateMutex);
	tracer.tracef(ARB, "StopSinkThread : entered sinkStateMutex\n");
	do
	{
		if (!bSinkThreadRunning)
		{
			tracer.tracef(EE, "StopSinkThread : bSinkThreadRunning = false\n");
			result = -10;
			waitForThread = false;
			break;
		}
		bSinkThreadShouldStop = true;
	}
	while (false);

	LeaveCriticalSection(&sinkStateMutex);
	tracer.tracef(ARB, "StopSinkThread : left  sinkStateMutex\n");
	if (waitForThread)
	{
		tracer.tracef(ARB, "StopSinkThread : before WaitForSingleObject\n");
#ifdef WIN32
		WaitForSingleObject(sinkThreadStopEvent, INFINITE);
#else
		sinkThreadStopEvent.wait(&sinkStateMutex, -1);
#endif
	}
	tracer.tracef(EE, "~StopSinkThread : returning %d\n", result);
	return result;
}

#ifdef WIN32
DWORD WINAPI
AudioSink::SinkThreadProc(LPVOID params)
#else
void*
AudioSink::SinkThreadProc(void* params)
#endif
{
	int result(0);
	int returnCode(0);
	AudioSink *audioSink = (AudioSink *)params;
	Tracer &tracer = audioSink->tracer;
	tracer.tracef(EE, "SinkThreadProc : begin\n");
	audioSink->bSinkThreadRunning = true;
	tracer.tracef(ARB, "SinkThreadProc : setting sinkThreadStartEvent\n");
#ifdef WIN32
	SetEvent(audioSink->sinkThreadStartEvent);
#else
	audioSink->sinkThreadStartEvent.signal();
#endif
	while (!audioSink->bSinkThreadShouldStop)
	{
		(audioSink->inputData).clear();
		tracer.tracef(DET, "SinkThreadProc : Entering audioSourcesMutex\n");
		EnterCriticalSection(&(audioSink->audioSourcesMutex));
		vector<AudioSource *>::iterator iter = (audioSink->audioSources).begin();
		for (iter = (audioSink->audioSources).begin(); iter != (audioSink->audioSources).end(); iter++)
		{
			AudioSample *pAudioSample;
			returnCode = (*iter)->GiveNextAudioSample(&pAudioSample, audioSink);
                        if(pAudioSample)
			(audioSink->inputData).push_back(make_pair(pAudioSample, *iter));
		}
		LeaveCriticalSection(&(audioSink->audioSourcesMutex));
		tracer.tracef(DET, "SinkThreadProc : Left audioSourcesMutex\n");
		tracer.tracef(DET, "SinkThreadProc : Rendering audioSamples\n");
		returnCode = audioSink->RenderAudioSamples(audioSink->inputData);
		for (int i=0; i<(audioSink->inputData).size(); i++)
		{
			AudioSample *sample = (audioSink->inputData[i]).first;
			if (sample)
			{
				sample->Release();
			}
		}
                vusleep(20000);
	}
	tracer.tracef(ARB, "SinkThreadProc : broke out of while loop\n");

	tracer.tracef(EE, "SinkThreadProc : entering sinkStateMutex\n");
	EnterCriticalSection(&(audioSink->sinkStateMutex));
	audioSink->bSinkThreadRunning = false;
	
	tracer.tracef(ARB, "SinkThreadProc : calling SinkThreadStopped\n");

#ifdef WIN32	
	audioSink->SinkThreadStopped(audioSink->sinkThreadHandle, audioSink->sinkThreadID);
	CloseHandle(audioSink->sinkThreadHandle);
	audioSink->sinkThreadHandle = NULL;
	audioSink->sinkThreadID = 0;
#else
        audioSink->sinkThreadHandle.exit();
#endif
	
	LeaveCriticalSection(&(audioSink->sinkStateMutex));
	tracer.tracef(EE, "SinkThreadProc : left sinkStateMutex\n");

	tracer.tracef(ARB, "SinkThreadProc : setting sinkThreadStopEvent and exiting\n");
#ifdef WIN32
	result = SetEvent(audioSink->sinkThreadStopEvent);
	return result;
#else
	result = audioSink->sinkThreadStopEvent.signal();
	return 0;
#endif
}

#ifdef WIN32
int
AudioSink::SinkThreadStarted(HANDLE sinkThreadHandle, DWORD sinkThreadID)
#else
int
AudioSink::SinkThreadStarted(vthread_t sinkThreadHandle, DWORD sinkThreadID)
#endif
{
	return 0;
}

#ifdef WIN32
int
AudioSink::SinkThreadStopped(HANDLE sinkThreadHandle, DWORD sinkThreadID)
#else
int
AudioSink::SinkThreadStopped(vthread_t sinkThreadHandle, DWORD sinkThreadID)
#endif
{
	return 0;
}