之前已经阅读了live555服务器的部分代码,其中也讲解了服务器对客户端各个RTSP命令的处理过程,下面我们来看看客户端是如何发送这些命令。testProgs中的OpenRTSP是典型的RTSPClient示例,所以分析它吧。
main()函数在playCommon.cpp文件中。main()的流程比较简单,跟服务端差别不大:建立任务计划对象--建立环境对象--处理用户输入的参数(RTSP地址)--创建RTSPClient实例--发出第一个RTSP请求(可能是OPTIONS也可能是DESCRIBE)--进入Loop。
RTSP的TCP连接是在发送第一个RTSP请求时才建立的,在RTSPClient的那几个发请求的函数sendXXXXXXCommand()中最终都调用sendRequest(),sendRequest()中会跟据情况建立起TCP连接。在建立连接时马上向任务计划中加入处理从这个TCP接收数据的socket handler:RTSPClient::incomingDataHandler()。
下面就是发送RTSP请求,OPTIONS就不必看了,从请求DESCRIBE开始:
void getSDPDescription(RTSPClient::responseHandler* afterFunc) { ourRTSPClient->sendDescribeCommand(afterFunc, ourAuthenticator); } unsigned RTSPClient::sendDescribeCommand(responseHandler* responseHandler, Authenticator* authenticator) { if (authenticator != NULL) fCurrentAuthenticator = *authenticator; return sendRequest(new RequestRecord(++fCSeq, "DESCRIBE", responseHandler)); }
参数responseHandler是调用者提供的回调函数,用于在处理完请求的回应后再调用之。并且在这个回调函数中会发出下一个请求--所有的请求都是这样依次发出的。使用回调函数的原因主要是因为socket的发送与接收不是同步进行的。类RequestRecord就代表一个请求,它不但保存了RTSP请求相关的信息,而且保存了请求完成后的回调函数--即responseHandler。有些请求发出时还没建立tcp连接,不能立即发送,则加入fRequestsAwaitingConnection队列;有些发出后要等待Server端的回应,就加入fRequestsAwaitingResponse队列,当收到回应后再从队列中把它取出。
下面我们就先来看看RTSPClient::sendRequest()里面的代码,其实无非是建立起RTSP请求字符串然后用TCP socket发送之。
unsigned RTSPClient::sendRequest(RequestRecord* request) { char* cmd = NULL; do { Boolean connectionIsPending = False; //判断客户端的请求是否需要等待 if (!fRequestsAwaitingConnection.isEmpty()) { // A connection is currently pending (with at least one enqueued request). Enqueue this request also: connectionIsPending = True; } //是否需要新打开一个连接 else if (fInputSocketNum < 0) { // we need to open a connection //这里面做了很多事情:解析地址->创建套接字,连接服务器->将socket加入fHandlers,循环获取状态并调用回调函数进行事件处理 int connectResult = openConnection(); if (connectResult < 0) break; // an error occurred else if (connectResult == 0) { // A connection is pending connectionIsPending = True; } // else the connection succeeded. Continue sending the command. } if (connectionIsPending) { fRequestsAwaitingConnection.enqueue(request); return request->cseq(); } // If requested (and we're not already doing it, or have done it), set up the special protocol for tunneling RTSP-over-HTTP: if (fTunnelOverHTTPPortNum != 0 && strcmp(request->commandName(), "GET") != 0 && fOutputSocketNum == fInputSocketNum) { if (!setupHTTPTunneling1()) break; fRequestsAwaitingHTTPTunneling.enqueue(request); return request->cseq(); } // Construct and send the command: // First, construct command-specific headers that we need: char* cmdURL = fBaseURL; // by default Boolean cmdURLWasAllocated = False; char const* protocolStr = "RTSP/1.0"; // by default char* extraHeaders = (char*)""; // by default Boolean extraHeadersWereAllocated = False; char* contentLengthHeader = (char*)""; // by default Boolean contentLengthHeaderWasAllocated = False; if (!setRequestFields(request, cmdURL, cmdURLWasAllocated, protocolStr, extraHeaders, extraHeadersWereAllocated)) { break; } char const* contentStr = request->contentStr(); // by default if (contentStr == NULL) contentStr = ""; unsigned contentStrLen = strlen(contentStr); if (contentStrLen > 0) { char const* contentLengthHeaderFmt = "Content-Length: %d\r\n"; unsigned contentLengthHeaderSize = strlen(contentLengthHeaderFmt) + 20 /* max int len */; contentLengthHeader = new char[contentLengthHeaderSize]; sprintf(contentLengthHeader, contentLengthHeaderFmt, contentStrLen); contentLengthHeaderWasAllocated = True; } //组包,发送RTSP命令 char* authenticatorStr = createAuthenticatorString(request->commandName(), fBaseURL); char const* const cmdFmt = "%s %s %s\r\n" "CSeq: %d\r\n" "%s" "%s" "%s" "%s" "\r\n" "%s"; unsigned cmdSize = strlen(cmdFmt) + strlen(request->commandName()) + strlen(cmdURL) + strlen(protocolStr) + 20 /* max int len */ + strlen(authenticatorStr) + fUserAgentHeaderStrLen + strlen(extraHeaders) + strlen(contentLengthHeader) + contentStrLen; cmd = new char[cmdSize]; sprintf(cmd, cmdFmt, request->commandName(), cmdURL, protocolStr, request->cseq(), authenticatorStr, fUserAgentHeaderStr, extraHeaders, contentLengthHeader, contentStr); delete[] authenticatorStr; if (cmdURLWasAllocated) delete[] cmdURL; if (extraHeadersWereAllocated) delete[] extraHeaders; if (contentLengthHeaderWasAllocated) delete[] contentLengthHeader; if (fVerbosityLevel >= 1) envir() << "Sending request: " << cmd << "\n"; if (fTunnelOverHTTPPortNum != 0 && strcmp(request->commandName(), "GET") != 0 && strcmp(request->commandName(), "POST") != 0) { // When we're tunneling RTSP-over-HTTP, we Base-64-encode the request before we send it. // (However, we don't do this for the HTTP "GET" and "POST" commands that we use to set up the tunnel.) char* origCmd = cmd; cmd = base64Encode(origCmd, strlen(cmd)); if (fVerbosityLevel >= 1) envir() << "\tThe request was base-64 encoded to: " << cmd << "\n\n"; delete[] origCmd; } if (send(fOutputSocketNum, cmd, strlen(cmd), 0) < 0) { char const* errFmt = "%s send() failed: "; unsigned const errLength = strlen(errFmt) + strlen(request->commandName()); char* err = new char[errLength]; sprintf(err, errFmt, request->commandName()); envir().setResultErrMsg(err); delete[] err; break; } // The command send succeeded, so enqueue the request record, so that its response (when it comes) can be handled. // However, note that we do not expect a response to a POST command with RTSP-over-HTTP, so don't enqueue that. int cseq = request->cseq(); if (fTunnelOverHTTPPortNum == 0 || strcmp(request->commandName(), "POST") != 0) { fRequestsAwaitingResponse.enqueue(request); } else { delete request; } delete[] cmd; return cseq; } while (0); // An error occurred, so call the response handler immediately (indicating the error): delete[] cmd; handleRequestError(request); delete request; return 0; }
接下来我们来看一下收到DESCRIBE的回应后如何处理它。理论上是跟据媒体信息建立起MediaSession了,看看是不是这样:
void continueAfterDESCRIBE(RTSPClient*, int resultCode, char* resultString) { if (resultCode != 0) { *env << "Failed to get a SDP description for the URL \"" << streamURL << "\": " << resultString << "\n"; delete[] resultString; shutdown(); } char* sdpDescription = resultString; *env << "Opened URL \"" << streamURL << "\", returning a SDP description:\n" << sdpDescription << "\n"; // Create a media session object from this SDP description: //根据服务器返回的SDP信息,创建一个会话 session = MediaSession::createNew(*env, sdpDescription); delete[] sdpDescription; if (session == NULL) { *env << "Failed to create a MediaSession object from the SDP description: " << env->getResultMsg() << "\n"; shutdown(); } else if (!session->hasSubsessions()) { *env << "This session has no media subsessions (i.e., no \"m=\" lines)\n"; shutdown(); } // Then, setup the "RTPSource"s for the session: MediaSubsessionIterator iter(*session); MediaSubsession *subsession; Boolean madeProgress = False; char const* singleMediumToTest = singleMedium; while ((subsession = iter.next()) != NULL) { // If we've asked to receive only a single medium, then check this now: if (singleMediumToTest != NULL) { if (strcmp(subsession->mediumName(), singleMediumToTest) != 0) { *env << "Ignoring \"" << subsession->mediumName() << "/" << subsession->codecName() << "\" subsession, because we've asked to receive a single " << singleMedium << " session only\n"; continue; } else { // Receive this subsession only singleMediumToTest = "xxxxx"; // this hack ensures that we get only 1 subsession of this type } } if (desiredPortNum != 0) { subsession->setClientPortNum(desiredPortNum); desiredPortNum += 2; } if (createReceivers) { //初始化subsession,在其中会建立RTP/RTCP socket以及RTPSource。 if (!subsession->initiate(simpleRTPoffsetArg)) { *env << "Unable to create receiver for \"" << subsession->mediumName() << "/" << subsession->codecName() << "\" subsession: " << env->getResultMsg() << "\n"; } else { *env << "Created receiver for \"" << subsession->mediumName() << "/" << subsession->codecName() << "\" subsession ("; if (subsession->rtcpIsMuxed()) { *env << "client port " << subsession->clientPortNum(); } else { *env << "client ports " << subsession->clientPortNum() << "-" << subsession->clientPortNum()+1; } *env << ")\n"; madeProgress = True; if (subsession->rtpSource() != NULL) { // Because we're saving the incoming data, rather than playing // it in real time, allow an especially large time threshold // (1 second) for reordering misordered incoming packets: unsigned const thresh = 1000000; // 1 second subsession->rtpSource()->setPacketReorderingThresholdTime(thresh); // Set the RTP source's OS socket buffer size as appropriate - either if we were explicitly asked (using -B), // or if the desired FileSink buffer size happens to be larger than the current OS socket buffer size. // (The latter case is a heuristic, on the assumption that if the user asked for a large FileSink buffer size, // then the input data rate may be large enough to justify increasing the OS socket buffer size also.) int socketNum = subsession->rtpSource()->RTPgs()->socketNum(); unsigned curBufferSize = getReceiveBufferSize(*env, socketNum); if (socketInputBufferSize > 0 || fileSinkBufferSize > curBufferSize) { unsigned newBufferSize = socketInputBufferSize > 0 ? socketInputBufferSize : fileSinkBufferSize; newBufferSize = setReceiveBufferTo(*env, socketNum, newBufferSize); if (socketInputBufferSize > 0) { // The user explicitly asked for the new socket buffer size; announce it: *env << "Changed socket receive buffer size for the \"" << subsession->mediumName() << "/" << subsession->codecName() << "\" subsession from " << curBufferSize << " to " << newBufferSize << " bytes\n"; } } } } } else { if (subsession->clientPortNum() == 0) { *env << "No client port was specified for the \"" << subsession->mediumName() << "/" << subsession->codecName() << "\" subsession. (Try adding the \"-p <portNum>\" option.)\n"; } else { madeProgress = True; } } } if (!madeProgress) shutdown(); //下一步就是发送SETUP请求了。需要为每个Track分别发送一次。 // Perform additional 'setup' on each subsession, before playing them: setupStreams(); }
上面的代码看起来非常长,实际上,去除一些打印,去除一些没必要的代码段之后是很短的,所以,希望大家还是可以把看阅读完。
getOptions(continueAfterOPTIONS);
|---sendOptionsCommand(afterFunc, ourAuthenticator);//1
.........|---sendRequest(new..RequestRecord(++fCSeq,"OPTIONS",responseHandler));//2
.................|---openConnection();//3
.........................|---connectToServer()//4
..................................|---connect()//5
..................................|---taskScheduler().setBackgroundHandling(connectionHandler);//6
.........................|---taskScheduler().setBackgroundHandling(incomingDataHandler);//7
.................|---send(fOutputSocketNum, cmd, strlen(cmd), 0)//8
.................|---fRequestsAwaitingResponse.enqueue(request);//9
2:RTSP第一步,发送OPTIONS包。并将continueAfterOPTIONS回调函数一同传入,通过这些参数构建一个RequestRecord对象,这些传入的参数通过这个对象管理
3:第一次使用RTSP协议首先要建立一个TCP连接, fInputSocketNum = fOutputSocketNum = setupStreamSocket(envir(), 0);先建立socket,然后调用4 connect连接。因为这个套接字并不是阻塞的,所以connect之后可能握手还没有完成就返回了。所以这是要添加connectionHandler到任务中,完成connect操作,一次就好。7:建立完成之后将接受消息处理函数添加到后台任务中。incomin–Data–Handler
8:前期准备都做好了然后发送OPTIONS命令,等待sever返回,
9:将开始生成的RequestRecord对象添加到等待队列中,其中包含了“OPTIONS”和处理函数“continueAfterXXXXX”
注:在发送OPTIONS时建立好了TCP连接,并添加好了处理函数
然后程序会进入singleEvent()循环,检查任务。
当检测到该socket有可读信息时,调用incomingDataHandler
在handleResponseBytes中先解析返回的信息,然后取出 fRequestsAwaitingResponse队列里的RequestRecord对象,上面已经说了,该对象保存了接受完OPTIONS之后的处理函数continueAfterOPTIONS。
然后进入 foundRequest->handler()
为了方便大家理解,我这里再把整个客户端的RTSP的DESCRIBE命令处理的流程整理一下:
1、客户端主动组包发送DESCRIBE命令,第一次命令发送会创建socket套接字,并设置数据处理回调函数
2、收到服务器响应后,通过回调函数的方式,客户端使用回调函数对响应报文做相应的处理,包括:根据SDP信息创建会话、为每个子会话创建RTP/RTCP套接字以及fRTPSource
3、为每个子会话,发送SETUP命令
这里先休息一下,我们继续往下讲剩余的几个RTSP命令!