rtpaudiosink.cxx

来自「rtp在linux下的实现」· CXX 代码 · 共 502 行

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

#include <stdio.h>
#include "RTPPacket.h"
#include "RTPAudioSink.h"
#include "AudioSample.h"

#ifndef WIN32
#include "NetworkAddress.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <unistd.h>
#endif

using namespace std;

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

RTPAudioSink::RTPAudioSink()
{
	char subFacilityName[100];
	sprintf(subFacilityName, "RTPAudioSink:%x", this);
	tracer.SetSubFacilityName(subFacilityName);
	SetTraceLevel();
	tracer.tracef(EE, "Constructor begin\n");
	rtpSocket = NULL;
	destinationIP = 0;
	destinationIPString = "127.0.0.1";
	destinationPort = 0;
	localPort = 0;
	tos = 0;
	bRunning = false;
	InitializeCriticalSection(&filterMutex);
#ifdef WIN32
	WSADATA wsaData;
	int result = WSAStartup(MAKEWORD(1,1), &wsaData);
	tracer.tracef(EE, "Constructor end : 0x%x\n", result);
#endif
}

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

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

int
RTPAudioSink::SetDestination(std::string ipAddress, unsigned short udpPort)
{
	tracer.tracef(EE, "SetDestination %s %u\n", ipAddress.c_str(), udpPort);
	int result = 0;
	destinationIPString = ipAddress;
	destinationPort = udpPort;

#ifdef WIN32
        destinationIP = inet_addr(ipAddress.c_str());
#else
        NetworkAddress na(ipAddress.c_str());
        destinationIP  = na.getIp4Address();
#endif

	if (!rtpSocket)
	{
		int result = CreateSocket();
	}
	if (result == 0)
	{
		sockaddr_in destinationAddr;
		memset((char *)&destinationAddr, 0, sizeof(destinationAddr));
		destinationAddr.sin_family = AF_INET;
		destinationAddr.sin_port = htons(destinationPort);

#ifdef WIN32
		destinationAddr.sin_addr.S_un.S_addr = destinationIP;
                fcntl( rtpSocket, F_SETFL, O_NDELAY);
		result = connect(rtpSocket, (sockaddr *)&destinationAddr, sizeof(destinationAddr));
		tracer.tracef(EE, "~SetDestination : %d 0x%x\n", result, result);
#else
                destinationAddr.sin_addr.s_addr = destinationIP;
		result = connect(rtpSocket, (sockaddr *)&destinationAddr, sizeof(destinationAddr));
#endif
	}
	return result;
}

int
RTPAudioSink::SetTOS(unsigned char tosByte)
{
	tracer.tracef(EE, "SetTOS 0x%x\n", tosByte);
	int result = 0;
#ifdef WIN32
	tos = tosByte;
	int tosValueInt = (int)tos;
	int tosLen = sizeof(int);
	if (rtpSocket)
	{
		tracer.tracef(SDI_LEVEL_ARBITRARY, "SetTOS : setsockopt\n");
		result = setsockopt(rtpSocket, IPPROTO_IP, IP_TOS, (char *)&tosValueInt, tosLen);
		if (result != 0)
		{
			result = WSAGetLastError();
			tracer.tracef(ERR, "SetTOS : e-setsockopt 0x%x\n", result);
		}
	}
	tracer.tracef(EE, "~SetTOS : %d 0x%x\n", result, result);
#endif
	return result;
}

int
RTPAudioSink::SetLocalPort(unsigned short port)
{
	tracer.tracef(EE, "SetLocalPort %u : localPort %u\n", port, localPort);
	int result(0);
	if (localPort != port)
	{
		localPort = port;
		if (rtpSocket)
		{
			result = CloseSocket();
			if (result == 0)
			{
				result = CreateSocket();
				if (result == 0)
				{
					result = SetDestination(destinationIPString, destinationPort);
				}
			}
		}
	}
	tracer.tracef(EE, "~SetLocalPort : %d 0x%x\n", result, result);
	return result;
}

int
RTPAudioSink::CreateSocket()
{
	tracer.tracef(EE, "CreateSocket\n"); 
	int result;

	if (rtpSocket)
	{
		result = CloseSocket();
	}

	rtpSocket = socket(AF_INET, SOCK_DGRAM, 0);

	sockaddr_in localSockAddr;
	memset((char *)&localSockAddr, 0, sizeof(localSockAddr));
	localSockAddr.sin_family = AF_INET;
	localSockAddr.sin_port = htons(localPort);
#ifdef WIN32
	localSockAddr.sin_addr.S_un.S_addr = INADDR_ANY;
	result = bind(rtpSocket, (sockaddr *)&localSockAddr, sizeof(localSockAddr));
	if (result != 0)
	{
		result = WSAGetLastError();
		tracer.tracef(ERR, "CreateSocket : e-bind : 0x%x\n", result);
	}
	result = setsockopt(rtpSocket, IPPROTO_IP, IP_TOS, (char *)&tos, 1);
	if (result != 0)
	{
		result = WSAGetLastError();
		tracer.tracef(ERR, "CreateSocket : e-setsockopt : 0x%x\n", result);
	}
#else
	localSockAddr.sin_addr.s_addr = INADDR_ANY;
	result = bind(rtpSocket, (sockaddr *)&localSockAddr, sizeof(localSockAddr));
        if(result != 0)
        {
        	tracer.tracef(ERR, "CreateSocket : failed : %s\n", strerror(errno));
        }
#endif
	tracer.tracef(EE, "~CreateSocket : %d 0x%x\n", result, result);
	return result;
}


int
RTPAudioSink::CloseSocket()
{
	tracer.tracef(EE, "CloseSocket\n");
	int result(0);
	if (rtpSocket)
	{
#ifdef WIN32
		tracer.tracef(ARB, "CloseSocket : closesocket\n");
		result = closesocket(rtpSocket);
		if (result != 0)
		{
			tracer.tracef(ERR, "CloseSocket : e-closesocket : 0x%x\n", WSAGetLastError());
			result = -10;
		}
#else
                close(rtpSocket);
#endif
	}
	rtpSocket = NULL;
	tracer.tracef(EE, "~CloseSocket : %d 0x%x\n", result, result);
	return 0;
}

int
RTPAudioSink::RenderAudioSamples(std::vector<std::pair<AudioSample*, AudioSource* > > &data)
{
	tracer.tracef(DET, "RenderAudioSamples\n");
        if(data.size() == 0)
        {
             tracer.tracef(DET, "No data to render\n");
             return 0;
        }
	AudioSample *sample = data[0].first;
	int dataSize = 0;
	int durationMillisec;

	char RTPPayloadType;
	bool transmitPacket = false;	// whether a packet will be transmitted
	bool setMarkerBit = false;

	int numBytes = sample->DataSize();
	durationMillisec = sample->GetDuration();
	if (durationMillisec <= 0)
	{
		durationMillisec = sample->GetSilenceDuration();
	}
	tracer.tracef(DET, "RenderAudioSamples : durationMillisec = %d\n", durationMillisec);

	timestamp += durationMillisec * 8;
	seqno = (seqno + 1) % 65536;

	if (numBytes > 0)
	{
		transmitPacket = true;
		setMarkerBit = silenceInterval;
		silenceInterval = false;

		tracer.tracef(DET, "RenderAudioSamples : memcpy(0x%x, 0x%x, %d)\n", sendBuffer+12, sample->Data(), sample->DataSize());
		memcpy(sendBuffer+12, sample->Data(), sample->DataSize());
		dataSize = 12 + sample->DataSize();

		WAVEFORMATEX wfx;
		sample->GetFormat(&wfx);
		switch(wfx.wFormatTag) 
		{
		case WAVE_FORMAT_MULAW		:	RTPPayloadType	= 0x00;
										break;
		case WAVE_FORMAT_ALAW		:	RTPPayloadType	= 0x08;
										break;
		case WAVE_FORMAT_CISCO_G729	:	RTPPayloadType	= 0x12;
										break;
		case WAVE_FORMAT_CISCO_G723_53:	RTPPayloadType	= 0x04;
										break;
		case WAVE_FORMAT_CISCO_G723_63:	RTPPayloadType	= 0x04;
										break;
		default						:	break;
		}
	}
	else if (!silenceInterval)
	{
		transmitPacket	= true;
		silenceInterval	= true;
		RTPPayloadType	= RTP_PAYLOADTYPE_SID;
		// See internet draft "RTP Payload for Comfort Noise". Actually 1 byte of data (although min amount
		// of data in RTP packet is 4 bytes. First byte is important. all others should be zero.
		// first byte = value between 0 to 127. If it is X, then noise level is -X dB. so 0 is loudest, and
		// 127 is quietest.
		// Pascal Huart in Dallas gave me this information
		*((long *)(sendBuffer + 12)) = htonl(0);
		sendBuffer[12] = 100;
		dataSize = 12 + 4;
	}

	if (transmitPacket)
	{
		sendBuffer[0] = 0x80;
		sendBuffer[1] = RTPPayloadType;
		if (setMarkerBit)
		{
			sendBuffer[1] |= 0x80;
		}
		*((short *)(sendBuffer + 2))	= htons((short)seqno);
		*((long  *)(sendBuffer + 4))	= htonl((long)timestamp);
		*((long  *)(sendBuffer + 8))	= htonl((long)ssrc);
		
		tracer.tracef(DET, "RenderAudioSamples : sending %d bytes\n", dataSize);
		send(rtpSocket, (char *)sendBuffer, dataSize, 0);
	}
	tracer.tracef(DET, "~RenderAudioSamples : seqno = %d, timeStamp = %d\n", seqno, timestamp);
	return 0;
}

int
RTPAudioSink::PrepareSink()
{
	int result(0);
	int returnCode(0);
	tracer.tracef(EE, "PrepareSink\n");
#ifdef WIN32
	srand((unsigned int)GetTickCount());
#endif
	ssrc = (unsigned long)rand();
	timestamp = (unsigned long)rand();
	seqno = (unsigned short)((unsigned long)(rand() % 65536));
	silenceInterval = true;		// so that the marker bit is set after the first packet
	tracer.tracef(EE, "PrepareSink : returning %d\n", result);
	return result;
}

int
RTPAudioSink::UnprepareSink()
{
	int result(0);
	int returnCode(0);
	tracer.tracef(EE, "UnprepareSink\n");
	tracer.tracef(EE, "UnprepareSink : returning %d\n", result);
	return result;
}

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

	LeaveCriticalSection(&filterMutex);
	tracer.tracef(ARB, "StartSink : left filterMutex\n");
	tracer.tracef(EE, "~StartSink : returning %d\n", result);
	return result;
}

int
RTPAudioSink::StopSink()
{
	int result(0);
	int returnCode(0);
	tracer.tracef(EE, "StopSink\n");
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "StopSink : entered filterMutex\n");
	do
	{
		if (!bRunning)
		{
			tracer.tracef(ERR, "StopSink : e-error : not running\n");
			result = -10;
			break;
		}
		returnCode = UnprepareSink();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StopSink : e-UnprepareSink : %d 0x%x\n", returnCode, returnCode);
			result = -20;
			break;
		}
		returnCode = AudioSink::StopSink();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "StopSink : e-AudioSink::StopSink : %d 0x%x\n", returnCode, returnCode);
			result = -30;
			break;
		}
		bRunning = false;
	}
	while (false);

	LeaveCriticalSection(&filterMutex);
	tracer.tracef(ARB, "StopSink : left filterMutex\n");
	tracer.tracef(EE, "~StopSink : returning %d\n", result);
	return result;
}

#ifdef WIN32
int
RTPAudioSink::SinkThreadStarted(HANDLE sinkThreadHandle, DWORD sinkThreadID)
#else
int
RTPAudioSink::SinkThreadStarted(vthread_t sinkThreadHandle, DWORD sinkThreadID)
#endif
{
	int returnCode(0);
	int result(0);
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "SinkThreadStarted\n");
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "SinkThreadStarted : entered filterMutex\n");
	do
	{
#ifdef WIN32

		returnCode = SetThreadPriority(sinkThreadHandle, THREAD_PRIORITY_TIME_CRITICAL);
		if (returnCode == 0)
		{
			returnCode = GetLastError();
			tracer.tracef(ERR, "SinkThreadStarted : e-SetThreadPriority : %d 0x%x\n", returnCode, returnCode);
		}
#endif
		returnCode = PrepareSink();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "SinkThreadStarted : e-PrepareSink : %d 0x%x\n", returnCode, returnCode);
			result = -10;
			break;
		}
		bRunning = true;
	}
	while(false);

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

#ifdef WIN32
int
RTPAudioSink::SinkThreadStopped(HANDLE sinkThreadHandle, DWORD sinkThreadID)
#else
int
RTPAudioSink::SinkThreadStopped(vthread_t sinkThreadHandle, DWORD sinkThreadID)
#endif
{
	tracer.tracef(SDI_LEVEL_ENTRY_EXIT, "SinkThreadStopped\n");
	int result(0);
	int returnCode(0);
	EnterCriticalSection(&filterMutex);
	tracer.tracef(ARB, "SinkThreadStopped : entered filterMutex\n");
	do
	{
		returnCode = UnprepareSink();
		if (returnCode != 0)
		{
			tracer.tracef(ERR, "SinkThreadStopped : e-UnprepareSink : %d 0x%x\n", returnCode, returnCode);
			result = -10;
			break;
		}
		bRunning = false;
	}
	while(false);

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