rtpaudiosource.cxx

rtp在linux下的实现

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

#ifndef WIN32
#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <unistd.h>
#endif
#include "RTPPacket.h"
#include "RTPAudioSource.h"
#include "AudioSample.h"
#include "AudioSampleManager.h"

using namespace std;

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

RTPAudioSource::RTPAudioSource()
{
	char subFacilityName[100];
	sprintf(subFacilityName, "RTPAudioSource:%x", this);
	tracer.SetSubFacilityName(subFacilityName);
	SetTraceLevel();
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "Constructor begin\n");
	rtpSocket = NULL;
	receivePort = 0;
	mode = PASSIVE_MODE;
	bRunning = false;
	InitializeCriticalSection(&filterMutex);
#ifdef WIN32
	WSADATA wsaData;
	int result = WSAStartup(MAKEWORD(1,1), &wsaData);
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "Constructor end : %d 0x%x\n", result, result);
#else
        CreateSocket();
#endif
}

RTPAudioSource::~RTPAudioSource()
{
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "Destructor begin\n");
	int result(0);
	result = CloseSocket();
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "RTPAudioSource~ : entered filterMutex\n");
	LeaveCriticalSection(&filterMutex);
	tracer.tracef(ARB, "RTPAudioSource~ : left filterMutex\n");
#ifdef WIN32
	DeleteCriticalSection(&filterMutex);
	result = WSACleanup();
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "Destructor end : %d 0x%x\n", result, result);
#endif
}

int
RTPAudioSource::SetTraceLevel()
{
#ifdef WIN32
	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", "RTPAudioSource", 0x100000);
	}
	tracer.SetSystemMask(SystemMask);
#endif
	return 0;
}

int
RTPAudioSource::PrepareSource()
{
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "PrepareSource\n");
	int result(0);
	if (!rtpSocket)
	{
		result = CreateSocket();
	}
	if (result == 0 && rtpSocket)
	{
		while(true)
		{
			// clear up all packets in the socket's input queue
			fd_set sdSet;
			FD_ZERO(&sdSet);
			FD_SET(rtpSocket, &sdSet);
			timeval timeOut;
			timeOut.tv_sec	= 0;
			timeOut.tv_usec	= 0;
			tracer.tracef(SDI_LEVEL_DETAILED, "PrepareSource : select\n");
			select(rtpSocket+1, &sdSet, NULL, NULL, &timeOut);
			if (FD_ISSET(rtpSocket, &sdSet))
			{
				tracer.tracef(SDI_LEVEL_DETAILED, "PrepareSource : recv\n");
				int bytesRead = recv(rtpSocket, receiveBuffer, RTPAUDIOSOURCE_RECEIVEBUFFER_SIZE, 0);			
				tracer.tracef(SDI_LEVEL_DETAILED, "PrepareSource : recv : got %d bytes\n", bytesRead);
				if (bytesRead <= 0)
				{
					break;
				}
			}
			else
			{
				break;
			}
		}
	}
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~PrepareSource : returning %d\n", result);
	return result;
}

int
RTPAudioSource::UnprepareSource()
{
	int result(0);
	int returnCode(0);
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "UnprepareSource\n");
	returnCode = CloseSocket();
	if (returnCode != 0)
	{
		tracer.tracef(ERR, "UnprepareSource : e-CloseSocket : %d 0x%x\n", returnCode, returnCode);
		result = -10;
	}
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~UnprepareSource : returning %d\n", result);
	return result;
}

int
RTPAudioSource::StartSource()
{
	SetTraceLevel();
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "StartSource\n");
	int result(0);
	int returnCode(0);
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "StartSource : entered filterMutex\n");
	do
	{
		if (bRunning)
		{
			tracer.tracef(ERR, "StartSource : e-error : already running\n");
			result = -10;
			break;
		}
		returnCode = PrepareSource();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StartSource : e-PrepareSource : %d 0x%x\n", returnCode, returnCode);
			result = -20;
			break;
		}
		returnCode = AudioSource::StartSource();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StartSource : e-AudioSource::StartSource : %d 0x%x\n", returnCode, returnCode);
			result = -30;
			break;
		}
		bRunning = true;
	}
	while(false);

	LeaveCriticalSection(&filterMutex);
	tracer.tracef(ARB, "StartSource : left filterMutex\n");
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~StartSource : %d 0x%x\n", result, result);
	return result;
}

int
RTPAudioSource::StopSource()
{
	int result(0);
	int returnCode(0);
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "StopSource\n");
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "StopSource : entered filterMutex\n");
	do
	{
		if (!bRunning)
		{
			tracer.tracef(ERR, "StopSource : e-error : not running\n");
			result = -10;
			break;
		}
		returnCode = UnprepareSource();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StopSource : e-CloseSocket : %d 0x%x\n", result, result);
			result = -20;
			break;
		}
		returnCode = AudioSource::StopSource();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StopSource : e-AudioSource::StopSource : %d 0x%x\n", returnCode, returnCode);
			result = -30;
			break;
		}
		bRunning = false;
		mode = PASSIVE_MODE;
	}
	while (false);

	LeaveCriticalSection(&filterMutex);
	tracer.tracef(ARB, "StopSource : left filterMutex\n");
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~StopSource : %d 0x%x\n", result, result);
	return result;
}

#ifdef WIN32
int
RTPAudioSource::SourceThreadStarted(HANDLE sourceThreadHandle, DWORD sourceThreadID)
#else
int
RTPAudioSource::SourceThreadStarted(vthread_t sourceThreadHandle, DWORD sourceThreadID)
#endif
{
	int returnCode(0);
	int result(0);
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "SourceThreadStarted\n");
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "SourceThreadStarted : entered filterMutex\n");
	do
	{
#ifdef WIN32
		returnCode = SetThreadPriority(sourceThreadHandle, THREAD_PRIORITY_TIME_CRITICAL);
		if (returnCode == 0)
		{
			returnCode = GetLastError();
			tracer.tracef(ERR, "SourceThreadStarted : e-SetThreadPriority : %d 0x%x\n", returnCode, returnCode);
		}
#endif
		returnCode = PrepareSource();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "SourceThreadStarted : e-PrepareSource : %d 0x%x\n", returnCode, returnCode);
			result = -10;
			break;
		}
		bRunning = true;
		mode = ACTIVE_MODE;
	}
	while(false);

	LeaveCriticalSection(&filterMutex);
	tracer.tracef(ARB, "SourceThreadStarted : left filterMutex\n");
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~SourceThreadStarted : %d 0x%x\n", result, result);
	return result;
}

#ifdef WIN32
int
RTPAudioSource::SourceThreadStopped(HANDLE sourceThreadHandle, DWORD sourceThreadID)
#else
int
RTPAudioSource::SourceThreadStopped(vthread_t sourceThreadHandle, DWORD sourceThreadID)
#endif
{
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "SourceThreadStopped\n");
	int result(0);
	int returnCode(0);
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "SourceThreadStopped : entered filterMutex\n");
	do
	{
		returnCode = UnprepareSource();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StopSourceThread : e-UnprepareSource : %d 0x%x\n", returnCode, returnCode);
			result = -10;
			break;
		}
		bRunning = false;
		mode = PASSIVE_MODE;
	}
	while(false);

	LeaveCriticalSection(&filterMutex);
	tracer.tracef(ARB, "SourceThreadStopped : left filterMutex\n");
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~SourceThreadStopped : %d 0x%x\n", result, result);
	return result;
}

int
RTPAudioSource::SetReceivePort(unsigned short receivePort)
{
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "SetReceivePort %u\n", receivePort);
	int result(0);
	this->receivePort = receivePort;
	if (rtpSocket)
	{
		sockaddr_in localSockAddr;
		memset((char *)&localSockAddr, 0, sizeof(localSockAddr));
		localSockAddr.sin_family = AF_INET;
		localSockAddr.sin_port = htons(receivePort);
		tracer.tracef(SDI_LEVEL_ARBITRARY, "SetReceivePort : calling bind\n");
#ifdef WIN32
		localSockAddr.sin_addr.S_un.S_addr = INADDR_ANY;
		result = bind(rtpSocket, (sockaddr *)&localSockAddr, sizeof(localSockAddr));
		if (result != 0)
		{
			tracer.tracef(ERR, "SetReceivePort : e-bind : 0x%x\n", WSAGetLastError());
		}
#else
		localSockAddr.sin_addr.s_addr = INADDR_ANY;
                result = bind(rtpSocket, (sockaddr *)&localSockAddr, sizeof(localSockAddr));
                if(result != 0)
                {
			tracer.tracef(ERR, "SetReceivePort : bind  failed: %s\n", strerror(errno));
                }
#endif
	}
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~SetReceivePort : %d 0x%x\n", result, result);
	return result;
}

int
RTPAudioSource::CreateSocket()
{
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "CreateSocket\n");
	int result(0);

	if (rtpSocket)
	{
		result = CloseSocket();
	}
	rtpSocket = socket(AF_INET, SOCK_DGRAM, 0);
#ifdef WIN32
	result = SetReceivePort(receivePort);
	LINGER linger;
	linger.l_onoff = 1;
	linger.l_linger = 0;
	BOOL dontLinger = 0;
	int sockoptValue = setsockopt(rtpSocket, SOL_SOCKET, SO_DONTLINGER, (char *)(&dontLinger), sizeof(BOOL));
	if (sockoptValue != 0)
	{
		sockoptValue = WSAGetLastError();
		tracer.tracef(ERR, "CreateSocket : e-setsockopt : %d\n", sockoptValue);
	}
#endif
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~CreateSocket : %d 0x%x\n", result, result);
	return result;
}

int
RTPAudioSource::CloseSocket()
{
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "CloseSocket\n");
	int result(0);
	if (rtpSocket)
	{
		tracer.tracef(SDI_LEVEL_ARBITRARY, "CloseSocket : calling shutdown\n");
		result = shutdown(rtpSocket, 0);	// SD_RECEIVE = 0, SD_SEND = 1, SD_BOTH = 2
		tracer.tracef(SDI_LEVEL_ARBITRARY, "CloseSocket : calling closesocket\n");
#ifdef WIN32
		result = closesocket(rtpSocket);
#else
		result = close(rtpSocket);
#endif
		rtpSocket = 0;
	}
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "~CloseSocket : %d 0x%x\n", result, result);
	return result;
}

int
RTPAudioSource::GenerateData(AudioSample **ppAudioSample)
{
	tracer.tracef(SDI_LEVEL_DETAILED, "RTPAudioSource::GenerateData\n");
	int result(0);
	AudioSample *audioSample;
	result = (AudioSampleManager::GetInstance())->GetAudioSample(&audioSample);
	*ppAudioSample = audioSample;
	if (audioSample)
	{
		audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_PCM_16_8_1));
		//audioSample->SetSilenceDuration(20);
		audioSample->SetSilenceDuration(0);
		audioSample->SetDataSize(0);
	}

	if (bRunning && rtpSocket && audioSample)
	{
		if (mode == ACTIVE_MODE)
		{
			// then this filter drives timing. so wait for packets
			tracer.tracef(SDI_LEVEL_DETAILED, "GenerateData : recv : active\n");
			result = recv(rtpSocket, receiveBuffer, RTPAUDIOSOURCE_RECEIVEBUFFER_SIZE, 0);
		}
		else
		{
			// passive mode means other filter somewhere is driving timing. in that case
			// do not delay. return immediately if there are no packets
			result = 0;
			fd_set sdSet;
			FD_ZERO(&sdSet);
			FD_SET(rtpSocket, &sdSet);
			timeval timeOut;
			timeOut.tv_sec	= 0;
			timeOut.tv_usec	= 0;
			tracer.tracef(SDI_LEVEL_DETAILED, "GenerateData : select\n");
			select(0, &sdSet, NULL, NULL, &timeOut);
			if (FD_ISSET(rtpSocket, &sdSet))
			{
				tracer.tracef(SDI_LEVEL_DETAILED, "GenerateData : recv : inactive\n");
				result = recv(rtpSocket, receiveBuffer, RTPAUDIOSOURCE_RECEIVEBUFFER_SIZE, 0);			
			}
		}
		tracer.tracef(SDI_LEVEL_DETAILED, "GenerateData : ~recv : %d\n", result);
		if (result > 0)
		{
			// actual bytes received
			RTPPacket *rtpPacket = new RTPPacket(receiveBuffer, result);
			if (rtpPacket && audioSample)
			{
				tracer.tracef(SDI_LEVEL_DETAILED, "GenerateData : memcpy(0x%x, 0x%x, %d)\n", audioSample->Data(), rtpPacket->RtpData(), rtpPacket->DataLen());
				memcpy(audioSample->Data(), rtpPacket->RtpData(), rtpPacket->DataLen());
				audioSample->SetDataSize(rtpPacket->DataLen());
				audioSample->SetSilenceDuration(0);
				audioSample->SetRTPHeader(rtpPacket);
				switch(rtpPacket->PayloadType())
				{
				case RTP_PAYLOADTYPE_G711_ULAW:
					audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_ULaw));
					break;
				case RTP_PAYLOADTYPE_G711_ALAW:
					audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_ALaw));
					break;
				case RTP_PAYLOADTYPE_G723:
					if (rtpPacket->DataLen() % 20 == 0)
					{
						audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_G723_53));
					}
					else
					{
						audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_G723_63));
					}
					break;
				case RTP_PAYLOADTYPE_G729:
					audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_G729));
					break;
				default:
					audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_PCM_16_8_1));
					break;
				}
			}
			else
			{
				tracer.tracef(SDI_LEVEL_ERROR, "GenerateData : e-malloc : rtpPacket = 0x%x, audioSample = 0x%x\n", rtpPacket, audioSample);
				result = -10;
			}
		}
		else
		{
			audioSample->SetDataSize(0);
			audioSample->SetSilenceDuration(20);
			audioSample->SetFormat(WaveFormat::GetWaveFormat(WaveFormat_PCM_16_8_1));
			result = 0;
#ifdef WIN32
			Sleep(20);
#else
			vusleep(20000);
#endif
		}
	}
	else
	{
		tracer.tracef(DET, "GenerateData : sleeping 20ms\n");
#ifdef WIN32
		Sleep(20);
#else
	        vusleep(20000);
#endif
	}
	tracer.tracef(SDI_LEVEL_DETAILED, "~GenerateData : %d\n", result);
	return result;
}