rtpaudiostream.cxx

rtp在linux下的实现

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

#include "RTPAudioStream.h"
#include "Parameters.h"
#include "AudioSample.h"
#include "AudioSampleManager.h"
#include "RTPPacket.h"

#define STATS_INTERVAL 100
int avgBufferSize(0);
int minBufferSize(0);
int numRepeats(0);
int numPackets(0);

using namespace std;

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

RTPAudioStream::RTPAudioStream(unsigned long ssrc)
{
	this->ssrc = ssrc;
	tracer = NULL;
	jitterBufferDepth = 0;
	restartThreshold = DEFAULT_JITTERBUFFER_SIZE;
	lastPacketTransmitted = NULL;
	lastOutputTimestamp = 0;
	lastOutputDuration = 0;
	lastPacketRepeatCount = 0;
	silenceIntervalInProgress = true;
	numSIDPackets = 0;
#ifdef WIN32
	dynamicJitterBuffer = GetRegKeyBool(HKEY_CURRENT_USER, "Software\\Cisco Systems\\CCNMediaTerm\\1.0", "UseDynamicJitterBuffer", true);
#else
        dynamicJitterBuffer = true;
#endif
	InitializeCriticalSection(&packetQueueMutex);
	InitializeCriticalSection(&audioStreamMutex);
}

RTPAudioStream::~RTPAudioStream()
{
	EnterCriticalSection(&packetQueueMutex);
	while (packetQueue.size() > 0)
	{
		AudioSample *audioSample;
		PopNextPacket(&audioSample);
		audioSample->Release();
	}
	LeaveCriticalSection(&packetQueueMutex);
	DeleteCriticalSection(&packetQueueMutex);
	DeleteCriticalSection(&audioStreamMutex);
}

int
RTPAudioStream::SetTracer(Tracer *tracer)
{
	this->tracer = tracer;
	return 0;
}

int
RTPAudioStream::SetRestartThreshold(int depthInMillisec)
{
	restartThreshold = depthInMillisec;
	return 0;
}

int
RTPAudioStream::NumQueuedPackets()
{
	return packetQueue.size();
}

int
RTPAudioStream::JitterBufferDepth()
{
	return jitterBufferDepth;
}

unsigned long
RTPAudioStream::SSRC()
{
	return ssrc;
}

int
RTPAudioStream::SetLastTransmittedPacket(AudioSample *newPacket)
{
	EnterCriticalSection(&audioStreamMutex);
	if (newPacket != lastPacketTransmitted)
	{
		if (lastPacketTransmitted)
		{
			lastPacketTransmitted->Release();
			lastPacketRepeatCount = 0;
		}
		lastPacketTransmitted = newPacket;
		if (newPacket)
		{
			newPacket->AddRef();
		}
	}
	LeaveCriticalSection(&audioStreamMutex);
	return 0;
}

int
RTPAudioStream::TimeDifference(unsigned long timestamp1, unsigned long timestamp2)
{
	int result(0);
	unsigned long timediff = timestamp1 > timestamp2 ? timestamp1 - timestamp2 : timestamp2 - timestamp1;
	if (timediff < 0x8fffffff)
	{
		result = timediff;
		return timestamp1 > timestamp2 ? -result : result;
	}
	else
	{
		unsigned long temp = (0xffffffff - timediff) + 1;
		result = temp;
		return timestamp1 > timestamp2 ? result : -result;
	}
}

int
RTPAudioStream::TimeDifference(AudioSample *sample1, AudioSample *sample2)
{
	return TimeDifference(sample1->RTPHeader()->Timestamp(), sample2->RTPHeader()->Timestamp());
}

int
RTPAudioStream::TimeDifferenceMillisec(AudioSample *sample1, AudioSample *sample2)
{
	int timestampDiff = TimeDifference(sample1, sample2);
	return (int)(((double)timestampDiff * sample1->MinFrameDuration())/1000.0 + 0.5);
}

int
RTPAudioStream::Clear()
{
	EnterCriticalSection(&audioStreamMutex);
	while(packetQueue.size() > 0)
	{
		AudioSample *audioSample;
		PopNextPacket(&audioSample);
		audioSample->Release();
	}
	silenceIntervalInProgress = true;
	SetLastTransmittedPacket(NULL);
	lastOutputTimestamp = 0;
	lastOutputDuration = 0;
	lastPacketRepeatCount = 0;
	numSIDPackets = 0;
	jitterBufferDepth = 0;
	numPackets = 0;
	numRepeats = 0;
#ifdef WIN32
	dynamicJitterBuffer = GetRegKeyBool(HKEY_CURRENT_USER, "Software\\Cisco Systems\\CCNMediaTerm\\1.0", "UseDynamicJitterBuffer", true);
#else
        dynamicJitterBuffer = true;
#endif
	LeaveCriticalSection(&audioStreamMutex);
	return 0;
}

int
RTPAudioStream::AddPacket(AudioSample *newPacket)
{
	EnterCriticalSection(&audioStreamMutex);
	RTPPacket *rtpHeader = newPacket->RTPHeader();
	if (jitterBufferDepth < restartThreshold * JITTERBUFFER_MAXSIZE_MULTIPLE)
	{
		if (rtpHeader->PayloadType() == RTP_PAYLOADTYPE_SID)
		{
			numSIDPackets++;
		}
		newPacket->AddRef();
		packetQueue.push(newPacket);
		jitterBufferDepth += newPacket->GetDuration();
	}
	LeaveCriticalSection(&audioStreamMutex);
	return 0;
}

int
RTPAudioStream::PopNextPacket(AudioSample **pAudioSample)
{
	if (packetQueue.size() > 0)
	{
		*pAudioSample = packetQueue.top();
		packetQueue.pop();
		RTPPacket *rtpHeader = (*pAudioSample)->RTPHeader();
		if (rtpHeader->PayloadType() == RTP_PAYLOADTYPE_SID)
		{
			numSIDPackets--;
		}
		else
		{
			jitterBufferDepth -= (*pAudioSample)->GetDuration();
		}
	}
	return 0;
}

int
RTPAudioStream::PopNextPacketForTransmit(AudioSample **pOutSample)
{
	AudioSample *audioSample;
	PopNextPacket(&audioSample);
	if (audioSample)
	{
		RTPPacket *rtpHeader = audioSample->RTPHeader();
		SetLastTransmittedPacket(audioSample);
		*pOutSample = audioSample;
		lastOutputTimestamp = rtpHeader->Timestamp();
		lastOutputDuration = audioSample->GetDuration();
		lastPacketRepeatCount = 1;
	}
	return 0;
}

int
RTPAudioStream::InsertRTPPacket(AudioSample *audioSample)
{
	RTPPacket *newPacketRTPHeader = audioSample->RTPHeader();
	EnterCriticalSection(&audioStreamMutex);
	if (lastPacketTransmitted && (TimeDifference(lastPacketTransmitted, audioSample) < 0))
	{
		RTPPacket *lastPacketRTPHeader = lastPacketTransmitted->RTPHeader();
		// if we receive a packet that is 25 packets before the last transmitted packet, it means a new stream
		if ((int)(newPacketRTPHeader->SeqNo()) - ((int)lastPacketRTPHeader->SeqNo()) < -25)
		{
			Clear();
			AddPacket(audioSample);
		}
	}
	else
	{
		AddPacket(audioSample);
	}
	LeaveCriticalSection(&audioStreamMutex);
	return 0;
}

int
RTPAudioStream::GiveNextRTPPacket(AudioSample **pOutSample)
{
	EnterCriticalSection(&audioStreamMutex);
	*pOutSample = NULL;
	if (silenceIntervalInProgress && (numSIDPackets > 0 || jitterBufferDepth >= restartThreshold))
	{
		silenceIntervalInProgress = false;
	}
	do
	{
		if (silenceIntervalInProgress)
		{
			break;
		}
		if (dynamicJitterBuffer && numPackets == 99)
		{
			tracer->tracef(EE, "GiveNextRTPPacket : numRepeats = %d\n", numRepeats);
			if (numRepeats <= 1)
			{
				int newThreshold(restartThreshold);
				if (jitterBufferDepth <= restartThreshold + 40)
				{
					newThreshold = restartThreshold - 20;
					if (newThreshold < 40) newThreshold = 40;
				}
				tracer->tracef(EE, "GiveNextRTPPacket : restartThreshold = %d, newThreshold = %d\n", restartThreshold, newThreshold);
				SetRestartThreshold(newThreshold);
				tracer->tracef(EE, "GiveNextRTPPacket : jitter buffer depth = %d\n", jitterBufferDepth);
				int durationDeleted(0);
				while(jitterBufferDepth > restartThreshold)
				{
					AudioSample *discardSample;
					PopNextPacketForTransmit(&discardSample);
					durationDeleted += discardSample->GetDuration();
					discardSample->Release();
					if (durationDeleted >= 60)
					{
						break;
					}
				}
				numRepeats = 0;
				numPackets = 100;
			}
			else
			{
				numRepeats = 0;
				restartThreshold = restartThreshold + 20;
				numPackets = 100;
				tracer->tracef(EE, "GiveNextRTPPacket : increased restartThreshold by 20 to %d\n", restartThreshold);
			}
		}
		if (packetQueue.size() <= 0)
		{
			if (lastPacketTransmitted && lastPacketRepeatCount < JITTERBUFFER_PLC_MAXREPEATS)
			{
				lastPacketTransmitted->AddRef();
				lastPacketRepeatCount++;
				if (lastPacketRepeatCount == 2)
				{
					numRepeats++;
				}
				lastOutputTimestamp += (lastOutputDuration * 1000) / lastPacketTransmitted->MinFrameDuration();
				*pOutSample = lastPacketTransmitted;
				numPackets = (numPackets + 1) % 100;
				break;
			}
			else
			{
				silenceIntervalInProgress = true;
				SetLastTransmittedPacket(NULL);
				break;
			}
			break;
		}

		AudioSample *audioSample = packetQueue.top();
		RTPPacket *rtpHeader = audioSample->RTPHeader();
		if (rtpHeader->PayloadType() == RTP_PAYLOADTYPE_SID)
		{
			PopNextPacket(&audioSample);
			audioSample->Release();
			silenceIntervalInProgress = true;
			SetLastTransmittedPacket(NULL);
			break;
		}
		if (!lastPacketTransmitted)
		{
			// transmit whichever is the next packet
			PopNextPacketForTransmit(pOutSample);
			numPackets = (numPackets + 1) % 100;
			break;
		}
		int timeDifference = TimeDifferenceMillisec(lastPacketTransmitted, audioSample) - lastOutputDuration * lastPacketRepeatCount;
		if (timeDifference < lastOutputDuration)
		{
			PopNextPacketForTransmit(pOutSample);
			numPackets = (numPackets + 1) % 100;
			break;
		}
		if (lastPacketRepeatCount < JITTERBUFFER_PLC_MAXREPEATS)
		{
			lastPacketTransmitted->AddRef();
			lastPacketRepeatCount++;
			lastOutputTimestamp += (lastOutputDuration * 1000) / lastPacketTransmitted->MinFrameDuration();
			*pOutSample = lastPacketTransmitted;
			numPackets = (numPackets + 1) % 100;
			break;
		}

		silenceIntervalInProgress = true;
		SetLastTransmittedPacket(NULL);

	} while(false);
	
	LeaveCriticalSection(&audioStreamMutex);
	return 0;
}