rtspserver.cpp

rtsp协议的主要实现代码.对开发流媒体

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

#include "stdafx.h"
#include "RTSPServer.h"
#include "Utility.h"
#include "time.h"

#if defined(__WIN32__) || defined(_WIN32) || defined(_QNX4)
#define _strncasecmp _strnicmp
#else
#define _strncasecmp strncasecmp
#endif

#define LISTEN_BACKLOG_SIZE 20
// By default, use INADDR_ANY for the sending and receiving interfaces:
netAddressBits SendingInterfaceAddr = INADDR_ANY;
netAddressBits ReceivingInterfaceAddr = INADDR_ANY;

typedef enum StreamingMode
{
	RTP_UDP,
		RTP_TCP,
		RAW_UDP
};

static void parseTransportHeader(char const* buf,
								 StreamingMode& streamingMode,
								 char*& streamingModeString,
								 char*& destinationAddressStr,
								 u_int8_t& destinationTTL,
								 portNumBits& clientRTPPortNum, // if UDP
								 portNumBits& clientRTCPPortNum, // if UDP
								 unsigned char& rtpChannelId, // if TCP
								 unsigned char& rtcpChannelId // if TCP
								 ) {
	// Initialize the result parameters to default values:
	streamingMode = RTP_UDP;
	streamingModeString = NULL;
	destinationAddressStr = NULL;
	destinationTTL = 255;
	clientRTPPortNum = 0;
	clientRTCPPortNum = 1; 
	rtpChannelId = rtcpChannelId = 0xFF;
	
	portNumBits p1, p2;
	unsigned ttl, rtpCid, rtcpCid;
	
	// First, find "Transport:"
	while (1) {
		if (*buf == '\0') return; // not found
		if (_strncasecmp(buf, "Transport: ", 11) == 0) break;
		++buf;
	}
	
	// Then, run through each of the fields, looking for ones we handle:
	char const* fields = buf + 11;
	char* field = strDupSize(fields);
	while (sscanf(fields, "%[^;]", field) == 1) {
		if (strcmp(field, "RTP/AVP/TCP") == 0) {
			streamingMode = RTP_TCP;
		} else if (strcmp(field, "RAW/RAW/UDP") == 0 ||
			strcmp(field, "MP2T/H2221/UDP") == 0) {
			streamingMode = RAW_UDP;
			streamingModeString = strDup(field);
		} else if (_strncasecmp(field, "destination=", 12) == 0) {
			delete[] destinationAddressStr;
			destinationAddressStr = strDup(field+12);
		} else if (sscanf(field, "ttl%u", &ttl) == 1) {
			destinationTTL = (u_int8_t)ttl;
		} else if (sscanf(field, "client_port=%hu-%hu", &p1, &p2) == 2) {
			clientRTPPortNum = p1;
			clientRTCPPortNum = p2;
		} else if (sscanf(field, "client_port=%hu", &p1) == 1) {
			clientRTPPortNum = p1;
			clientRTCPPortNum = streamingMode == RAW_UDP ? 0 : p1 + 1;
		} else if (sscanf(field, "interleaved=%u-%u", &rtpCid, &rtcpCid) == 2) {
			rtpChannelId = (unsigned char)rtpCid;
			rtcpChannelId = (unsigned char)rtcpCid;
		}
		
		fields += strlen(field);
		while (*fields == ';') ++fields; // skip over separating ';' chars
		if (*fields == '\0' || *fields == '\r' || *fields == '\n') break;
	}
	delete[] field;
}

static Boolean parseRangeHeader(char const* buf, float& rangeStart, float& rangeEnd) {
	// Initialize the result parameters to default values:
	rangeStart = rangeEnd = 0.0;
	
	// First, find "Range:"
	while (1) {
		if (*buf == '\0') return False; // not found
		if (_strncasecmp(buf, "Range: ", 7) == 0) break;
		++buf;
	}
	
	// Then, run through each of the fields, looking for ones we handle:
	char const* fields = buf + 7;
	while (*fields == ' ') ++fields;
	float start, end;
	if (sscanf(fields, "npt = %f - %f", &start, &end) == 2) {
		rangeStart = start;
		rangeEnd = end;
	} else if (sscanf(fields, "npt = %f -", &start) == 1) {
		rangeStart = start;
	} else {
		return False; // The header is malformed
	}
	
	return True;
}

static Boolean parseScaleHeader(char const* buf, float& scale) {
	// Initialize the result parameter to a default value:
	scale = 1.0;
	
	// First, find "Scale:"
	while (1) {
		if (*buf == '\0') return False; // not found
		if (_strncasecmp(buf, "Scale: ", 7) == 0) break;
		++buf;
	}
	
	// Then, run through each of the fields, looking for ones we handle:
	char const* fields = buf + 7;
	while (*fields == ' ') ++fields;
	float sc;
	if (sscanf(fields, "%f", &sc) == 1) {
		scale = sc;
	} else {
		return False; // The header is malformed
	}
	
	return True;
}

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////
RTSPServer::RTSPServer(Port ourPort) :
	 fServerPort(ourPort),fSessionIdCounter(0)
{
		 int		iResult; 
		 WSADATA ws; 
		 iResult = WSAStartup(0x0101,&ws); 
		 if(iResult)
			 return ;
}

RTSPServer::~RTSPServer()
{

}

DWORD WINAPI incomingConnectionHandler(LPVOID lpParam)
{
	RTSPServer* server = (RTSPServer*)lpParam;
	while(1)
	{
		server->incomingConnectionHandler1();
	}
	return 0;
}

void RTSPServer::RunServer()
{
	fServerSocket = fBankSocket.Create();
	if(!fServerSocket)
		return ;

	struct sockaddr_in name;
    name.sin_family = AF_INET;
    name.sin_port = fServerPort.num();
    name.sin_addr.s_addr = 0;
    if (bind(fServerSocket, (struct sockaddr*)&name, sizeof name) != 0) 
	{
		char tmpBuffer[100];
		sprintf(tmpBuffer, "bind() error (port number: %d): ",
			ntohs(fServerPort.num()));
		closesocket(fServerSocket);
		return;
	}
	// Allow multiple simultaneous connections:
    if (listen(fServerSocket, LISTEN_BACKLOG_SIZE) < 0) 
	{
		CUtility::setResultErrMsg("listen() failed: ");
		return ;
    }	

	CreateThread(NULL,0,incomingConnectionHandler,this,0,NULL);
	fBasicTaskScheduler.doEventLoop(); //need thread
}

void RTSPServer::incomingConnectionHandler1()
{
	struct sockaddr_in clientAddr;
	SOCKLEN_T clientAddrLen = sizeof clientAddr;
	int clientSocket = accept(fServerSocket, (struct sockaddr*)&clientAddr,
		&clientAddrLen);
	
	if(clientSocket < 0)
		return ;
	
	// Create a new object for this RTSP session:
	new RTSPClientSession(*this, ++fSessionIdCounter,
		clientSocket, clientAddr);
	return ;
}

////////// RTSPServer::RTSPClientSession //////////
#define PARAM_STRING_MAX 100

void RTSPServer::RTSPClientSession
::incomingRequestHandler(void* instance, int /*mask*/) 
{
	RTSPClientSession* Session = (RTSPClientSession*)instance;
	Session->incomingRequestHandler1();
}

RTSPServer::RTSPClientSession
::~RTSPClientSession()
{
	// Turn off background read handling:
	fOurServer.fBasicTaskScheduler.turnOffBackgroundReadHandling(fClientSocket);
	::closesocket(fClientSocket);
	reclaimStreamStates();

	//fOurServer.removeServerMediaSession(fOurServerMediaSession);
}

RTSPServer::RTSPClientSession
::RTSPClientSession(RTSPServer& ourServer, unsigned sessionId,
					int clientSocket, struct sockaddr_in clientAddr)
					: fOurServer(ourServer), fOurSessionId(sessionId),
					fOurServerMediaSession(NULL),fStreamAfterSETUP(False),
					fClientSocket(clientSocket), fClientAddr(clientAddr),				
					fNumStreamStates(0), fStreamStates(NULL),fSessionIsActive(True)
{
	fOurServer.fBasicTaskScheduler.turnOnBackgroundReadHandling(clientSocket,(BackgroundHandlerProc*)&RTSPClientSession::incomingRequestHandler,this);
}

void RTSPServer::RTSPClientSession::incomingRequestHandler1()
{
	//struct sockaddr_in dummy; // 'from' address, meaningless in this case
	int bytesLeft = sizeof fBuffer;
	int totalBytes = 0;
	Boolean endOfMsg = False;
	unsigned char* ptr = fBuffer;
	unsigned char* lastCRLF = ptr-3;
	
	while (!endOfMsg) {
		if (bytesLeft <= 0) {
			// command too big
			delete this;
			return;
		}
		
		int bytesRead = recv(fClientSocket,(char *)ptr,bytesLeft,0);

		if (bytesRead <= 0) {
			// The client socket has apparently died - kill it:
			//delete this;
			return;
		}
#ifdef DEBUG
		ptr[bytesRead] = '\0';
		fprintf(stderr, "RTSPClientSession[%p]::incomingRequestHandler1() read %d bytes:\n%s\n", this, bytesRead, ptr);
#endif
		
		// Look for the end of the message: <CR><LF><CR><LF>
		unsigned char *tmpPtr = ptr;
		if (totalBytes > 0) --tmpPtr; // In case the last read ended with a <CR>
		while (tmpPtr < &ptr[bytesRead-1]) {
			if (*tmpPtr == '\r' && *(tmpPtr+1) == '\n') {
				if (tmpPtr - lastCRLF == 2) { // This is it:
					endOfMsg = 1;
					break;
				}
				lastCRLF = tmpPtr;
			}
			++tmpPtr;
		}
		
		bytesLeft -= bytesRead;
		totalBytes += bytesRead;
		ptr += bytesRead;
	}
	fBuffer[totalBytes] = '\0';
	
	// Parse the request string into command name and 'CSeq',
	// then handle the command:
	char cmdName[PARAM_STRING_MAX];
	char urlPreSuffix[PARAM_STRING_MAX];
	char urlSuffix[PARAM_STRING_MAX];
	char cseq[PARAM_STRING_MAX];
	if (!parseRequestString((char*)fBuffer, totalBytes,
		cmdName, sizeof cmdName,
		urlPreSuffix, sizeof urlPreSuffix,
		urlSuffix, sizeof urlSuffix,
		cseq, sizeof cseq)) {
#ifdef DEBUG
		fprintf(stderr, "parseRequestString() failed!\n");
#endif
		handleCmd_bad(cseq);
	} else {
#ifdef DEBUG
		fprintf(stderr, "parseRequestString() returned cmdName \"%s\", urlPreSuffix \"%s\", urlSuffix \"%s\"\n", cmdName, urlPreSuffix, urlSuffix);
#endif
		if (strcmp(cmdName, "OPTIONS") == 0) {
			handleCmd_OPTIONS(cseq);
		} else if (strcmp(cmdName, "DESCRIBE") == 0) {
			handleCmd_DESCRIBE(cseq, urlSuffix, (char const*)fBuffer);
		} else if (strcmp(cmdName, "SETUP") == 0) {
			handleCmd_SETUP(cseq, urlPreSuffix, urlSuffix, (char const*)fBuffer);
		} else if (strcmp(cmdName, "TEARDOWN") == 0
			|| strcmp(cmdName, "PLAY") == 0
			|| strcmp(cmdName, "PAUSE") == 0
			|| strcmp(cmdName, "GET_PARAMETER") == 0) {
			handleCmd_withinSession(cmdName, urlPreSuffix, urlSuffix, cseq,
				(char const*)fBuffer);
		} else {
			handleCmd_notSupported(cseq);
		}
	}
    
#ifdef DEBUG
	fprintf(stderr, "sending response: %s\n", fResponseBuffer);
#endif
	send(fClientSocket, (char const*)fResponseBuffer, strlen((char*)fResponseBuffer), 0);

	if (!fSessionIsActive)
	{
		delete this;
	}
}

Boolean
RTSPServer::RTSPClientSession
  ::parseRequestString(char const* reqStr,
		       unsigned reqStrSize,
		       char* resultCmdName,
		       unsigned resultCmdNameMaxSize,
		       char* resultURLPreSuffix,
		       unsigned resultURLPreSuffixMaxSize,
		       char* resultURLSuffix,