audiosource.cxx

rtp在linux下的实现

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

#include <algorithm>
#include <iostream>

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

using namespace std;

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

AudioSource::AudioSource()
{
	bSourceRunning = false;
	bSourceThreadRunning = false;
	bSourceThreadShouldStop = true;
#ifdef WIN32
	sourceThreadHandle = NULL;
	sourceThreadID = 0;
	sourceThreadStartEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
	sourceThreadStopEvent  = CreateEvent(NULL, FALSE, FALSE, NULL);
	InitializeCriticalSection(&audioSinksMutex);
	InitializeCriticalSection(&sourceThreadMutex);
	InitializeCriticalSection(&sourceStateMutex);
#endif
}

AudioSource::~AudioSource()
{
#ifdef WIN32
	CloseHandle(sourceThreadStartEvent);
	CloseHandle(sourceThreadStopEvent);
	DeleteCriticalSection(&audioSinksMutex);
	DeleteCriticalSection(&sourceThreadMutex);
	DeleteCriticalSection(&sourceStateMutex);
#endif
}

int 
AudioSource::SubscribeAudioSink(AudioSink *audioSink)
{
	EnterCriticalSection(&audioSinksMutex);
	if (find(audioSinks.begin(), audioSinks.end(), audioSink) == audioSinks.end())
	{
		audioSinks.push_back(audioSink);
	}
	LeaveCriticalSection(&audioSinksMutex);
	return 0;
}

int
AudioSource::UnsubscribeAudioSink(AudioSink *audioSink)
{
	EnterCriticalSection(&audioSinksMutex);
	vector<AudioSink *>::iterator position = NULL;
	position = find(audioSinks.begin(), audioSinks.end(), audioSink);
	if (position != audioSinks.end())
	{
		audioSinks.erase(position);
	}
	LeaveCriticalSection(&audioSinksMutex);
	return 0;
}

int
AudioSource::StartSource()
{
	int result(0);
	EnterCriticalSection(&sourceStateMutex);
	if (bSourceThreadRunning || bSourceRunning)
	{
		result = -10;
	}
	else
	{
		bSourceRunning = true;
	}
	LeaveCriticalSection(&sourceStateMutex);
	return result;
}

int
AudioSource::StopSource()
{
	int result(0);
	EnterCriticalSection(&sourceStateMutex);
	if (bSourceThreadRunning)
	{
		result = -10;
	}
	else
	{
		bSourceRunning = false;
	}
	LeaveCriticalSection(&sourceStateMutex);
	return result;
}

int
AudioSource::StartSourceThread()
{
	tracer.tracef(EE, "StartSourceThread\n");
	int result(0);
	
	tracer.tracef(ARB, "StartSourceThread : entering  sourceStateMutex\n");
	EnterCriticalSection(&sourceStateMutex);
	do
	{
		if (bSourceRunning || bSourceThreadRunning)
		{
			tracer.tracef(ERR, "StartSourceThread : e-error : bSourceRunning = %d, bSourceThreadRunning = %d\n", bSourceRunning, bSourceThreadRunning);
			result = -10;
			break;
		}
#ifdef WIN32
		if (sourceThreadHandle != NULL || sourceThreadID != 0)
		{
			tracer.tracef(ERR, "StartSourceThread : e-error : sourceThreadHandle = 0x%x, sourceThreadID = 0x%x\n", sourceThreadHandle, sourceThreadID);
			result = -20;
			break;
		}
		BOOL resetResult;
		tracer.tracef(ARB, "StartSourceThread : resetting sourceThreadStartEvent and sourceThreadStopEvent\n");
		resetResult = ResetEvent(sourceThreadStartEvent);
		if (resetResult == 0)
		{
			tracer.tracef(ERR, "StartSourceThread : e-ResetEvent(sourceThreadStartedEvent) : 0x%x\n", GetLastError());
		}
		resetResult = ResetEvent(sourceThreadStopEvent);
		if (resetResult == 0)
		{
			tracer.tracef(ERR, "StartSourceThread : e-ResetEvent(sourceThreadStopEvent) : 0x%x\n", GetLastError());
		}
#endif
		bSourceThreadShouldStop = true;		// we make it false only if everything else succeeds
#ifdef WIN32
		sourceThreadHandle = CreateThread(NULL, 0, SourceThreadProc, (void *)this, CREATE_SUSPENDED, &sourceThreadID);
		if (sourceThreadHandle == NULL)
		{
			result = -20;
			break;
		}
		result = SourceThreadStarted(sourceThreadHandle, sourceThreadID);
		if (result != 0)
		{
			result = -30;
			break;
		}
		bSourceThreadShouldStop = false;
		DWORD result = ResumeThread(sourceThreadHandle);
		if (result == -1)
		{
			result = -40;
			break;
		}
		tracer.tracef(ARB, "StartSourceThread : before WaitForSingleObject(sourceThreadStartEvent, INFINITE)\n");
		DWORD waitResult = WaitForSingleObject(sourceThreadStartEvent, INFINITE);
		tracer.tracef(ARB, "StartSourceThread : waitResult = 0x%x\n", waitResult);
		if (waitResult == WAIT_FAILED)
		{
			waitResult = GetLastError();
			tracer.tracef(ERR, "StartSourceThread : e-WaitForSingleObject : %d 0x%x\n", waitResult, waitResult);
		}
#else
		bSourceThreadShouldStop = false;
                sourceThreadHandle.spawn(SourceThreadProc, (void *)this);
                sourceThreadStartEvent.wait(&sourceStateMutex, -1);
		SourceThreadStarted(*sourceThreadHandle.getId(), *sourceThreadHandle.getId());
#endif
	} while (false);
	LeaveCriticalSection(&sourceStateMutex);
	tracer.tracef(ARB, "StartSourceThread : left  sourceStateMutex\n");
	tracer.tracef(EE, "~StartSourceThread : returning %d\n", result);
	return result;
}

int
AudioSource::StopSourceThread()
{
	int result(0);
	bool waitForThread = true;
	tracer.tracef(EE, "StopSourceThread\n");
	tracer.tracef(ARB, "StopSourceThread : entering  sourceStateMutex\n");
	EnterCriticalSection(&sourceStateMutex);
	do
	{
		if (!bSourceThreadRunning)
		{
			tracer.tracef(EE, "StopSourceThread : bSourceThreadRunning = false\n");
			result =  -10;
			waitForThread = false;
			break;
		}
		bSourceThreadShouldStop = true;
	} while (false);

	LeaveCriticalSection(&sourceStateMutex);
	tracer.tracef(ARB, "StopSourceThread : left  sourceStateMutex\n");
	if (waitForThread)
	{
		tracer.tracef(ARB, "StopSourceThread : before WaitForSingleObject(sourceThreadStopEvent, INFINITE)\n");
#ifdef WIN32
		DWORD waitResult = WaitForSingleObject(sourceThreadStopEvent, INFINITE);
		tracer.tracef(ARB, "StopSourceThread : waitResult = 0x%x\n", waitResult);
		if (waitResult == WAIT_FAILED)
		{
			waitResult = GetLastError();
			tracer.tracef(ERR, "StopSourceThread : e-WaitForSingleObject : %d 0x%x\n", waitResult, waitResult);
		}
#else
		sourceThreadStopEvent.wait(&sourceStateMutex, -1);
#endif
	}
	tracer.tracef(EE, "~StopSourceThread : returning %d\n", result);
	return result;
}

#ifdef WIN32

DWORD WINAPI
AudioSource::SourceThreadProc(LPVOID params)
#else
void* 
AudioSource::SourceThreadProc(void* params)
#endif
{
	int result(0);
	AudioSource *audioSource = (AudioSource *)params;
	audioSource->tracer.tracef(EE, "SourceThreadProc : begin\n");
	audioSource->bSourceThreadRunning = true;
#ifdef WIN32
	SetEvent(audioSource->sourceThreadStartEvent);
#else
	audioSource->sourceThreadStartEvent.signal();
#endif
	while (!audioSource->bSourceThreadShouldStop)
	{
		AudioSample *pAudioSample;
		pAudioSample = NULL;
		if ((result = audioSource->GenerateData(&pAudioSample)) < 0)
		{
			break;
		}
		if (pAudioSample)
		{
			EnterCriticalSection(&(audioSource->audioSinksMutex));
			vector<AudioSink *>::iterator iter = (audioSource->audioSinks).begin();
			for (iter = (audioSource->audioSinks).begin(); iter != (audioSource->audioSinks).end(); iter++)
			{
				int i = audioSource->SendAudioSample(pAudioSample, *iter);
			}
			LeaveCriticalSection(&(audioSource->audioSinksMutex));
			pAudioSample->Release();
		}
	}
	audioSource->tracer.tracef(ARB, "SourceThreadProc : broke out of while loop. bSourceThreadShouldStop = %d\n", audioSource->bSourceThreadShouldStop);

	audioSource->tracer.tracef(EE, "SourceThreadProc : entering sourceStateMutex\n");
	EnterCriticalSection(&(audioSource->sourceStateMutex));
	audioSource->bSourceThreadRunning = false;
	
	audioSource->tracer.tracef(ARB, "SourceThreadProc : calling SourceThreadStopped\n");
#ifdef WIN32
	audioSource->SourceThreadStopped(audioSource->sourceThreadHandle, audioSource->sourceThreadID);
	CloseHandle(audioSource->sourceThreadHandle);
	audioSource->sourceThreadHandle = NULL;
	audioSource->sourceThreadID = 0;
#else
        audioSource->sourceThreadHandle.exit();
#endif

	LeaveCriticalSection(&(audioSource->sourceStateMutex));
	audioSource->tracer.tracef(EE, "SourceThreadProc : left sourceStateMutex\n");

	audioSource->tracer.tracef(EE, "SourceThreadProc : setting sourceThreadStopEvent and exiting\n");
#ifdef WIN32
	BOOL setResult = SetEvent(audioSource->sourceThreadStopEvent);
#else
	audioSource->sourceThreadStopEvent.signal();
#endif
	return 0;
}

int
AudioSource::SendAudioSample(AudioSample *audioSample, AudioSink *audioSink)
{
	tracer.tracef(DET, "SendAudioSample : calling TakeNextAudioSample\n");
	return audioSink->TakeNextAudioSample(audioSample, this);
}

#ifdef WIN32
int
AudioSource::SourceThreadStarted(HANDLE sourceThreadHandle, DWORD sourceThreadID)
#else
int
AudioSource::SourceThreadStarted(vthread_t sourceThreadHandle, DWORD sourceThreadID)
#endif
{
	return 0;
}

#ifdef WIN32
int
AudioSource::SourceThreadStopped(HANDLE sourceThreadHandle, DWORD sourceThreadID)
#else
int
AudioSource::SourceThreadStopped(vthread_t sourceThreadHandle, DWORD sourceThreadID)
#endif
{
	return 0;
}

int
AudioSource::GiveNextAudioSample(AudioSample **ppAudioSample, AudioSink *audioSink)
{
	int result = GenerateData(ppAudioSample);
	EnterCriticalSection(&audioSinksMutex);
	vector<AudioSink *>::iterator iter = audioSinks.begin();
	for (iter = audioSinks.begin(); iter != audioSinks.end(); iter++)
	{
		if (*(iter) == audioSink)
		{
			continue;
		}
		SendAudioSample(*ppAudioSample, *iter);
	}
	LeaveCriticalSection(&audioSinksMutex);
	return 0;
}

int
AudioSource::GenerateData(AudioSample **audioSample)
{
	tracer.tracef(ERR, "GenerateData : ERROR : Pure Virtual Function Call !!!\n");
	return 0;
}