求索阁

Live555不仅实现了RTSP服务器端，还实现了RTSP客户端，我们通过testRTSPClient.cpp这个程序来看一下，Live555的RTSP客户端与服务器端建立RTSP连接的过程。

首先来看一下main函数：

1 char eventLoopWatchVariable = 0;

2

3 int main(int argc, char** argv) {

4   // Begin by setting up our usage environment:

5   TaskScheduler* scheduler = BasicTaskScheduler::createNew();

6   UsageEnvironment* env = BasicUsageEnvironment::createNew(*scheduler);

7

8   // We need at least one "rtsp://" URL argument:

9   if (argc < 2) {

10     usage(*env, argv[0]);

11     return 1;

12   }

13

14   // There are argc-1 URLs: argv[1] through argv[argc-1].  Open and start streaming each one:

15   for (int i = 1; i <= argc-1; ++i) {

16     openURL(*env, argv[0], argv[i]);

17   }

18

19   // All subsequent activity takes place within the event loop:

20   env->taskScheduler().doEventLoop(&eventLoopWatchVariable);

21     // This function call does not return, unless, at some point in time, "eventLoopWatchVariable" gets set to something non-zero.

22

23   return 0;

24

25   // If you choose to continue the application past this point (i.e., if you comment out the "return 0;" statement above),

26   // and if you don't intend to do anything more with the "TaskScheduler" and "UsageEnvironment" objects,

27   // then you can also reclaim the (small) memory used by these objects by uncommenting the following code:

28   /*

29     env->reclaim(); env = NULL;

30     delete scheduler; scheduler = NULL;

31   */

32 }

和testOnDeamandRTSPServer.cpp一样，首先也是创建TaskScheduler对象和UsageEnvironment对象，然后调用openURL函数去请求某个媒体资源，参数是该媒体资源的RTSP地址，最后使程序进入主循环。

1 void openURL(UsageEnvironment& env, char const* progName, char const* rtspURL) {

2   // Begin by creating a "RTSPClient" object.  Note that there is a separate "RTSPClient" object for each stream that we wish

3   // to receive (even if more than stream uses the same "rtsp://" URL).

4   RTSPClient* rtspClient = ourRTSPClient::createNew(env, rtspURL, RTSP_CLIENT_VERBOSITY_LEVEL, progName);

5   if (rtspClient == NULL) {

6     env << "Failed to create a RTSP client for URL \"" << rtspURL << "\": " << env.getResultMsg() << "\n";

7     return;

8   }

9

10   ++rtspClientCount;

11

12   // Next, send a RTSP "DESCRIBE" command, to get a SDP description for the stream.

13   // Note that this command - like all RTSP commands - is sent asynchronously; we do not block, waiting for a response.

14   // Instead, the following function call returns immediately, and we handle the RTSP response later, from within the event loop:

15   rtspClient->sendDescribeCommand(continueAfterDESCRIBE); //发送DESCRIBE命令，并传入回调函数

16 }

OpenURL函数很简单，创建一个RTSPClient对象，一个RTSPClient对象代表一个RTSP客户端。然后调用sendDescribeCommand函数发送DESCRIBE命令，回调函数是continueAfterDESCRIBE函数，在收到RTSP服务器端对DESCRIBE命令的回复时调用。

1 void continueAfterDESCRIBE(RTSPClient* rtspClient, int resultCode, char* resultString) {

2   do {

3     UsageEnvironment& env = rtspClient->envir(); // alias

4     StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias

5

6     if (resultCode != 0) {  // 返回结果码非0表示出错

7       env << *rtspClient << "Failed to get a SDP description: " << resultString << "\n";

8       delete[] resultString;

9       break;

10     }

11     // resultString即从服务器端返回的SDP信息字符串

12     char* const sdpDescription = resultString;

13     env << *rtspClient << "Got a SDP description:\n" << sdpDescription << "\n";

14

15     // Create a media session object from this SDP description:

16     scs.session = MediaSession::createNew(env, sdpDescription);   //根据SDP信息创建一个MediaSession对象

17     delete[] sdpDescription; // because we don't need it anymore

18     if (scs.session == NULL) {

19       env << *rtspClient << "Failed to create a MediaSession object from the SDP description: " << env.getResultMsg() << "\n";

20       break;

21     } else if (!scs.session->hasSubsessions()) {

22       env << *rtspClient << "This session has no media subsessions (i.e., no \"m=\" lines)\n";

23       break;

24     }

25

26     // Then, create and set up our data source objects for the session.  We do this by iterating over the session's 'subsessions',

27     // calling "MediaSubsession::initiate()", and then sending a RTSP "SETUP" command, on each one.

28     // (Each 'subsession' will have its own data source.)

29     scs.iter = new MediaSubsessionIterator(*scs.session);

30     setupNextSubsession(rtspClient); //开始对服务器端的每个ServerMediaSubsession发送SETUP命令请求建立连接

31     return;

32   } while (0);

33

34   // An unrecoverable error occurred with this stream.

35   shutdownStream(rtspClient);

36 }

客户端收到服务器端对DESCRIBE命令的回复，取得SDP信息后，客户端创建一个MediaSession对象。MediaSession和ServerMediaSession是相对应的概念，MediaSession表示客户端请求服务器端某个媒体资源的会话，类似地，客户端还存在与ServerMediaSubsession相对应的MediaSubsession，表示MediaSession的子会话，创建MediaSession的同时也创建了包含的MediaSubsession对象。然后客户端对服务器端的每个ServerMediaSubsession发送SETUP命令请求建立连接。

1 void setupNextSubsession(RTSPClient* rtspClient) {

2   UsageEnvironment& env = rtspClient->envir(); // alias

3   StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias

4  

5   scs.subsession = scs.iter->next();

6   if (scs.subsession != NULL) {

7     if (!scs.subsession->initiate()) {  // 调用initiate函数初始化MediaSubsession对象

8       env << *rtspClient << "Failed to initiate the \"" << *scs.subsession << "\" subsession: " << env.getResultMsg() << "\n";

9       setupNextSubsession(rtspClient); // give up on this subsession; go to the next one

10     } else {

11       env << *rtspClient << "Initiated the \"" << *scs.subsession << "\" subsession (";

12       if (scs.subsession->rtcpIsMuxed()) {

13     env << "client port " << scs.subsession->clientPortNum();

14       } else {

15     env << "client ports " << scs.subsession->clientPortNum() << "-" << scs.subsession->clientPortNum()+1;

16       }

17       env << ")\n";

18       // 发送SETUP命令

19       // Continue setting up this subsession, by sending a RTSP "SETUP" command:

20       rtspClient->sendSetupCommand(*scs.subsession, continueAfterSETUP, False, REQUEST_STREAMING_OVER_TCP);

21     }

22     return;

23   }

24   // 成功与所有的ServerMediaSubsession建立了连接，现在发送PLAY命令

25   // We've finished setting up all of the subsessions.  Now, send a RTSP "PLAY" command to start the streaming:

26   if (scs.session->absStartTime() != NULL) {

27     // Special case: The stream is indexed by 'absolute' time, so send an appropriate "PLAY" command:

28     rtspClient->sendPlayCommand(*scs.session, continueAfterPLAY, scs.session->absStartTime(), scs.session->absEndTime());

29   } else {

30     scs.duration = scs.session->playEndTime() - scs.session->playStartTime();

31     rtspClient->sendPlayCommand(*scs.session, continueAfterPLAY);

32   }

33 }

34

35 void continueAfterSETUP(RTSPClient* rtspClient, int resultCode, char* resultString) {

36   do {

37     UsageEnvironment& env = rtspClient->envir(); // alias

38     StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias

39

40     if (resultCode != 0) {

41       env << *rtspClient << "Failed to set up the \"" << *scs.subsession << "\" subsession: " << resultString << "\n";

42       break;

43     }

44

45     env << *rtspClient << "Set up the \"" << *scs.subsession << "\" subsession (";

46     if (scs.subsession->rtcpIsMuxed()) {

47       env << "client port " << scs.subsession->clientPortNum();

48     } else {

49       env << "client ports " << scs.subsession->clientPortNum() << "-" << scs.subsession->clientPortNum()+1;

50     }

51     env << ")\n";

52

53     // Having successfully setup the subsession, create a data sink for it, and call "startPlaying()" on it.

54     // (This will prepare the data sink to receive data; the actual flow of data from the client won't start happening until later,

55     // after we've sent a RTSP "PLAY" command.)

56     //对每个MediaSubsession创建一个MediaSink对象来请求和保存数据

57     scs.subsession->sink = DummySink::createNew(env, *scs.subsession, rtspClient->url());

58       // perhaps use your own custom "MediaSink" subclass instead

59     if (scs.subsession->sink == NULL) {

60       env << *rtspClient << "Failed to create a data sink for the \"" << *scs.subsession

61       << "\" subsession: " << env.getResultMsg() << "\n";

62       break;

63     }

64

65     env << *rtspClient << "Created a data sink for the \"" << *scs.subsession << "\" subsession\n";

66     scs.subsession->miscPtr = rtspClient; // a hack to let subsession handle functions get the "RTSPClient" from the subsession

67     scs.subsession->sink->startPlaying(*(scs.subsession->readSource()),

68                        subsessionAfterPlaying, scs.subsession);          // 调用MediaSink的startPlaying函数准备播放

69     // Also set a handler to be called if a RTCP "BYE" arrives for this subsession:

70     if (scs.subsession->rtcpInstance() != NULL) {

71       scs.subsession->rtcpInstance()->setByeHandler(subsessionByeHandler, scs.subsession);

72     }

73   } while (0);

74   delete[] resultString;

75

76   // Set up the next subsession, if any:  与下一个ServerMediaSubsession建立连接

77   setupNextSubsession(rtspClient);

78 }

setupNextSubsession函数中首先调用MediaSubsession的initiate函数初始化MediaSubsession，然后对ServerMediaSubsession发送SETUP命令，收到回复后回调continueAfterSETUP函数。在continueAfterSETUP函数中，为MediaSubsession创建MediaSink对象来请求和保存服务器端发送的数据，然后调用MediaSink::startPlaying函数开始准备播放对应的ServerMediaSubsession，最后调用setupNextSubsession函数与下一个ServerMediaSubsession建立连接，在setupNextSubsession函数中，会检查是否与所有的ServerMediaSubsession都建立了连接，是则发送PLAY命令请求开始传送数据，收到回复则调用continueAfterPLAY函数。

在客户端发送PLAY命令之前，我们先看一下MediaSubsession::initiate函数的内容：

  1 Boolean MediaSubsession::initiate(int useSpecialRTPoffset) {

  2   if (fReadSource != NULL) return True; // has already been initiated

  3

  4   do {

  5     if (fCodecName == NULL) {

  6       env().setResultMsg("Codec is unspecified");

  7       break;

  8     }

  9     //创建客户端socket，包括RTP socket和RTCP socket，准备从服务器端接收数据

10     // Create RTP and RTCP 'Groupsocks' on which to receive incoming data.

11     // (Groupsocks will work even for unicast addresses)

12     struct in_addr tempAddr;

13     tempAddr.s_addr = connectionEndpointAddress();

14         // This could get changed later, as a result of a RTSP "SETUP"

15     //使用指定的RTP端口和RTCP端口，RTP端口必须是偶数，而RTCP端口必须是（RTP端口+1）

16     if (fClientPortNum != 0 && (honorSDPPortChoice || IsMulticastAddress(tempAddr.s_addr))) {

17       // The sockets' port numbers were specified for us.  Use these:

18       Boolean const protocolIsRTP = strcmp(fProtocolName, "RTP") == 0;

19       if (protocolIsRTP && !fMultiplexRTCPWithRTP) {

20     fClientPortNum = fClientPortNum&~1;

21         // use an even-numbered port for RTP, and the next (odd-numbered) port for RTCP

22       }

23       if (isSSM()) {

24     fRTPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, fClientPortNum);

25       } else {

26     fRTPSocket = new Groupsock(env(), tempAddr, fClientPortNum, 255);

27       }

28       if (fRTPSocket == NULL) {

29     env().setResultMsg("Failed to create RTP socket");

30     break;

31       }

32      

33       if (protocolIsRTP) {

34     if (fMultiplexRTCPWithRTP) {

35       // Use the RTP 'groupsock' object for RTCP as well:

36       fRTCPSocket = fRTPSocket;

37     } else {

38       // Set our RTCP port to be the RTP port + 1:

39       portNumBits const rtcpPortNum = fClientPortNum|1;

40       if (isSSM()) {

41         fRTCPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, rtcpPortNum);

42       } else {

43         fRTCPSocket = new Groupsock(env(), tempAddr, rtcpPortNum, 255);

44       }

45     }

46       }

47     } else {

          // 选取随机的RTP端口和RTCP端口

48       // Port numbers were not specified in advance, so we use ephemeral port numbers.

49       // Create sockets until we get a port-number pair (even: RTP; even+1: RTCP).

50       // (However, if we're multiplexing RTCP with RTP, then we create only one socket,

51       // and the port number  can be even or odd.)

52       // We need to make sure that we don't keep trying to use the same bad port numbers over

53       // and over again, so we store bad sockets in a table, and delete them all when we're done.

54       HashTable* socketHashTable = HashTable::create(ONE_WORD_HASH_KEYS);

55       if (socketHashTable == NULL) break;

56       Boolean success = False;

57       NoReuse dummy(env());

58           // ensures that our new ephemeral port number won't be one that's already in use

59

60       while (1) {

61     // Create a new socket:

62     if (isSSM()) {

63       fRTPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, 0);

64     } else {

65       fRTPSocket = new Groupsock(env(), tempAddr, 0, 255);

66     }

67     if (fRTPSocket == NULL) {

68       env().setResultMsg("MediaSession::initiate(): unable to create RTP and RTCP sockets");

69       break;

70     }

71

72     // Get the client port number:

73     Port clientPort(0);

74     if (!getSourcePort(env(), fRTPSocket->socketNum(), clientPort)) {

75       break;

76     }

77     fClientPortNum = ntohs(clientPort.num());

78

79     if (fMultiplexRTCPWithRTP) {

80       // Use this RTP 'groupsock' object for RTCP as well:

81       fRTCPSocket = fRTPSocket;

82       success = True;

83       break;

84     }     

85

86     // To be usable for RTP, the client port number must be even:

87     if ((fClientPortNum&1) != 0) { // it's odd

88       // Record this socket in our table, and keep trying:

89       unsigned key = (unsigned)fClientPortNum;

90       Groupsock* existing = (Groupsock*)socketHashTable->Add((char const*)key, fRTPSocket);

91       delete existing; // in case it wasn't NULL

92       continue;

93     }

94

95     // Make sure we can use the next (i.e., odd) port number, for RTCP:

96     portNumBits rtcpPortNum = fClientPortNum|1;

97     if (isSSM()) {

98       fRTCPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, rtcpPortNum);

99     } else {

100       fRTCPSocket = new Groupsock(env(), tempAddr, rtcpPortNum, 255);

101     }

102     if (fRTCPSocket != NULL && fRTCPSocket->socketNum() >= 0) {

103       // Success! Use these two sockets.

104       success = True;

105       break;

106     } else {

107       // We couldn't create the RTCP socket (perhaps that port number's already in use elsewhere?).

108       delete fRTCPSocket; fRTCPSocket = NULL;

109

110       // Record the first socket in our table, and keep trying:

111       unsigned key = (unsigned)fClientPortNum;

112       Groupsock* existing = (Groupsock*)socketHashTable->Add((char const*)key, fRTPSocket);

113       delete existing; // in case it wasn't NULL

114       continue;

115     }

116       }

117

118       // Clean up the socket hash table (and contents):

119       Groupsock* oldGS;

120       while ((oldGS = (Groupsock*)socketHashTable->RemoveNext()) != NULL) {

121     delete oldGS;

122       }

123       delete socketHashTable;

124

125       if (!success) break; // a fatal error occurred trying to create the RTP and RTCP sockets; we can't continue

126     }

127

128     // Try to use a big receive buffer for RTP - at least 0.1 second of

129     // specified bandwidth and at least 50 KB

130     unsigned rtpBufSize = fBandwidth * 25 / 2; // 1 kbps * 0.1 s = 12.5 bytes

131     if (rtpBufSize < 50 * 1024)

132       rtpBufSize = 50 * 1024;

133     increaseReceiveBufferTo(env(), fRTPSocket->socketNum(), rtpBufSize);

134

135     if (isSSM() && fRTCPSocket != NULL) {

136       // Special case for RTCP SSM: Send RTCP packets back to the source via unicast:

137       fRTCPSocket->changeDestinationParameters(fSourceFilterAddr,0,~0);

138     }

139     //创建FramedSource对象来请求数据

140     // Create "fRTPSource" and "fReadSource":

141     if (!createSourceObjects(useSpecialRTPoffset)) break;

142

143     if (fReadSource == NULL) {

144       env().setResultMsg("Failed to create read source");

145       break;

146     }

147     // 创建RTCPInstance对象

148     // Finally, create our RTCP instance. (It starts running automatically)

149     if (fRTPSource != NULL && fRTCPSocket != NULL) {

150       // If bandwidth is specified, use it and add 5% for RTCP overhead.

151       // Otherwise make a guess at 500 kbps.

152       unsigned totSessionBandwidth

153     = fBandwidth ? fBandwidth + fBandwidth / 20 : 500;

154       fRTCPInstance = RTCPInstance::createNew(env(), fRTCPSocket,

155                           totSessionBandwidth,

156                           (unsigned char const*)

157                           fParent.CNAME(),

158                           NULL /* we're a client */,

159                           fRTPSource);

160       if (fRTCPInstance == NULL) {

161     env().setResultMsg("Failed to create RTCP instance");

162     break;

163       }

164     }

165

166     return True;

167   } while (0);

168

169   deInitiate();

170   fClientPortNum = 0;

171   return False;

172 }

在MediaSubsession::initiate函数中，首先创建了两个客户端socket分别用于接收RTP数据和RTCP数据；然后创建FramedSource对象用来从服务器端请求数据，FramedSource对象在createSourceObjects函数中被创建，createSourceObjects根据ServerMediaSubsession资源的不同格式创建不同的FramedSource，我们还是以H264视频为例，则创建的是H264VideoRTPSource对象；最后还创建了RTCPInstance对象。

接下来，我们继续看客户端收到PLAY命令回复后调用continueAfterPLAY函数：

1 void continueAfterPLAY(RTSPClient* rtspClient, int resultCode, char* resultString) {

2   Boolean success = False;

3

4   do {

5     UsageEnvironment& env = rtspClient->envir(); // alias

6     StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias

7

8     if (resultCode != 0) {

9       env << *rtspClient << "Failed to start playing session: " << resultString << "\n";

10       break;

11     }

12

13     // Set a timer to be handled at the end of the stream's expected duration (if the stream does not already signal its end

14     // using a RTCP "BYE").  This is optional.  If, instead, you want to keep the stream active - e.g., so you can later

15     // 'seek' back within it and do another RTSP "PLAY" - then you can omit this code.

16     // (Alternatively, if you don't want to receive the entire stream, you could set this timer for some shorter value.)

17     if (scs.duration > 0) {

18       unsigned const delaySlop = 2; // number of seconds extra to delay, after the stream's expected duration.  (This is optional.)

19       scs.duration += delaySlop;

20       unsigned uSecsToDelay = (unsigned)(scs.duration*1000000);

21       scs.streamTimerTask = env.taskScheduler().scheduleDelayedTask(uSecsToDelay, (TaskFunc*)streamTimerHandler, rtspClient);

22     }

23

24     env << *rtspClient << "Started playing session";

25     if (scs.duration > 0) {

26       env << " (for up to " << scs.duration << " seconds)";

27     }

28     env << "...\n";

29

30     success = True;

31   } while (0);

32   delete[] resultString;

33

34   if (!success) {

35     // An unrecoverable error occurred with this stream.

36     shutdownStream(rtspClient);

37   }

38 }

continueAfterPLAY函数的内容很简单，只是简单地打印出“Started  playing  session”。在服务器端收到PLAY命令后，就开始向客户端发送RTP数据包和RTCP数据包，而客户端在MediaSink::startPlaying函数中就开始等待接收来自服务器端的视频数据。

在continueAfterSETUP函数中创建的MediaSink是DummySink对象，DummySink是MediaSink的子类，这个例子中客户端没有利用收到的视频数据，所以叫做DummySink。

客户端调用MediaSink::startPlaying函数开始接收服务器端的数据，这个函数和之前介绍服务器端建立RTSP连接过程时是同一个函数

1 Boolean MediaSink::startPlaying(MediaSource& source,

2                 afterPlayingFunc* afterFunc,

3                 void* afterClientData) {

4   // Make sure we're not already being played:

5   if (fSource != NULL) {

6     envir().setResultMsg("This sink is already being played");

7     return False;

8   }

9

10   // Make sure our source is compatible:

11   if (!sourceIsCompatibleWithUs(source)) {

12     envir().setResultMsg("MediaSink::startPlaying(): source is not compatible!");

13     return False;

14   }

15   fSource = (FramedSource*)&source;       //此处的fSource是之前创立的H264VideoRTPSource对象

16

17   fAfterFunc = afterFunc;

18   fAfterClientData = afterClientData;

19   return continuePlaying();

20 }

在MediaSink::startPlaying函数中又调用DummySink::continuePlaying函数

1 Boolean DummySink::continuePlaying() {

2   if (fSource == NULL) return False; // sanity check (should not happen)

3

4   // Request the next frame of data from our input source.  "afterGettingFrame()" will get called later, when it arrives:

5   fSource->getNextFrame(fReceiveBuffer, DUMMY_SINK_RECEIVE_BUFFER_SIZE,

6                         afterGettingFrame, this,

7                         onSourceClosure, this);

8   return True;

9 }

在DummySink::continuePlaying函数中通过H264VideoRTPSource对象请求服务器端的数据，H264VideoRTPSource是MultiFramedRTPSource的子类，请求成功后回调DummySink::afterGettingFrame函数。在FramedSource::getNextFrame函数中，调用了MultiFramedRTPSource::doGetNextFrame函数：

1 void MultiFramedRTPSource::doGetNextFrame() {

2   if (!fAreDoingNetworkReads) {

3     // Turn on background read handling of incoming packets:

4     fAreDoingNetworkReads = True;

5     TaskScheduler::BackgroundHandlerProc* handler

6       = (TaskScheduler::BackgroundHandlerProc*)&networkReadHandler;

7     fRTPInterface.startNetworkReading(handler); //通过RTPInterface对象读取网络数据，在服务器端是通过RTPInterface对象发送网络数据

//读到数据后回调networkReadHandler函数来处理

8   }

9

10   fSavedTo = fTo; //读到的数据保存在fTo中

11   fSavedMaxSize = fMaxSize;

12   fFrameSize = 0; // for now

13   fNeedDelivery = True;

14   doGetNextFrame1();

15 }

16

17 void MultiFramedRTPSource::doGetNextFrame1() {

18   while (fNeedDelivery) { 19     // If we already have packet data available, then deliver it now.

20     Boolean packetLossPrecededThis;

21     BufferedPacket* nextPacket

22       = fReorderingBuffer->getNextCompletedPacket(packetLossPrecededThis);

23     if (nextPacket == NULL) break;

24

25     fNeedDelivery = False;

26

27     if (nextPacket->useCount() == 0) {

28       // Before using the packet, check whether it has a special header

29       // that needs to be processed:

30       unsigned specialHeaderSize;

31       if (!processSpecialHeader(nextPacket, specialHeaderSize)) {

32     // Something's wrong with the header; reject the packet:

33     fReorderingBuffer->releaseUsedPacket(nextPacket);

34     fNeedDelivery = True;

35     break;

36       }

37       nextPacket->skip(specialHeaderSize);

38     }

39

40     // Check whether we're part of a multi-packet frame, and whether

41     // there was packet loss that would render this packet unusable:

42     if (fCurrentPacketBeginsFrame) {

43       if (packetLossPrecededThis || fPacketLossInFragmentedFrame) {

44     // We didn't get all of the previous frame.

45     // Forget any data that we used from it:

46     fTo = fSavedTo; fMaxSize = fSavedMaxSize;

47     fFrameSize = 0;

48       }

49       fPacketLossInFragmentedFrame = False;

50     } else if (packetLossPrecededThis) {

51       // We're in a multi-packet frame, with preceding packet loss

52       fPacketLossInFragmentedFrame = True;

53     }

54     if (fPacketLossInFragmentedFrame) {

55       // This packet is unusable; reject it:

56       fReorderingBuffer->releaseUsedPacket(nextPacket);

57       fNeedDelivery = True;

58       break;

59     }

60

61     // The packet is usable. Deliver all or part of it to our caller:

62     unsigned frameSize;

63     nextPacket->use(fTo, fMaxSize, frameSize, fNumTruncatedBytes,

64             fCurPacketRTPSeqNum, fCurPacketRTPTimestamp,

65             fPresentationTime, fCurPacketHasBeenSynchronizedUsingRTCP,

66             fCurPacketMarkerBit);

67     fFrameSize += frameSize;

68

69     if (!nextPacket->hasUsableData()) {

70       // We're completely done with this packet now

71       fReorderingBuffer->releaseUsedPacket(nextPacket);

72     }

73

74     if (fCurrentPacketCompletesFrame) {         // 成功读到一帧数据

75       // We have all the data that the client wants.

76       if (fNumTruncatedBytes > 0) {

77     envir() << "MultiFramedRTPSource::doGetNextFrame1(): The total received frame size exceeds the client's buffer size ("

78         << fSavedMaxSize << ").  "

79         << fNumTruncatedBytes << " bytes of trailing data will be dropped!\n";

80       }

81       // Call our own 'after getting' function, so that the downstream object can consume the data:

82       if (fReorderingBuffer->isEmpty()) {

83     // Common case optimization: There are no more queued incoming packets, so this code will not get

84     // executed again without having first returned to the event loop.  Call our 'after getting' function

85     // directly, because there's no risk of a long chain of recursion (and thus stack overflow):

86     afterGetting(this);

87       } else {

88     // Special case: Call our 'after getting' function via the event loop.

89     nextTask() = envir().taskScheduler().scheduleDelayedTask(0,

90                                  (TaskFunc*)FramedSource::afterGetting, this);

91       }

92     } else {

93       // This packet contained fragmented data, and does not complete

94       // the data that the client wants.  Keep getting data:

95       fTo += frameSize; fMaxSize -= frameSize;

96       fNeedDelivery = True;

97     }

98   }

99 }

  在doGetNextFrame1函数中，若成功读取到一个完整的帧，则调用Framed::afterGetting函数，进一步回调了DummySink::afterGettingFrame函数

1 void DummySink::afterGettingFrame(void* clientData, unsigned frameSize, unsigned numTruncatedBytes,

2                   struct timeval presentationTime, unsigned durationInMicroseconds) {

3   DummySink* sink = (DummySink*)clientData;

4   sink->afterGettingFrame(frameSize, numTruncatedBytes, presentationTime, durationInMicroseconds);

5 }

6

7 // If you don't want to see debugging output for each received frame, then comment out the following line:

8 #define DEBUG_PRINT_EACH_RECEIVED_FRAME 1

9

10 void DummySink::afterGettingFrame(unsigned frameSize, unsigned numTruncatedBytes,

11                   struct timeval presentationTime, unsigned /*durationInMicroseconds*/) {

12   // We've just received a frame of data.  (Optionally) print out information about it:

13 #ifdef DEBUG_PRINT_EACH_RECEIVED_FRAME

14   if (fStreamId != NULL) envir() << "Stream \"" << fStreamId << "\"; ";

15   envir() << fSubsession.mediumName() << "/" << fSubsession.codecName() << ":\tReceived " << frameSize << " bytes";

16   if (numTruncatedBytes > 0) envir() << " (with " << numTruncatedBytes << " bytes truncated)";

17   char uSecsStr[6+1]; // used to output the 'microseconds' part of the presentation time

18   sprintf(uSecsStr, "%06u", (unsigned)presentationTime.tv_usec);

19   envir() << ".\tPresentation time: " << (int)presentationTime.tv_sec << "." << uSecsStr;

20   if (fSubsession.rtpSource() != NULL && !fSubsession.rtpSource()->hasBeenSynchronizedUsingRTCP()) {

21     envir() << "!"; // mark the debugging output to indicate that this presentation time is not RTCP-synchronized

22   }

23 #ifdef DEBUG_PRINT_NPT

24   envir() << "\tNPT: " << fSubsession.getNormalPlayTime(presentationTime);

25 #endif

26   envir() << "\n";

27 #endif

28  

29   // Then continue, to request the next frame of data:

30   continuePlaying();

31 }

在DummySink::afterGettingFrame函数中只是简单地打印出了某个MediaSubsession接收到了多少字节的数据，然后接着利用FramedSource去读取数据。可以看出，在RTSP客户端，Live555也是在MediaSink和FramedSource之间形成了一个循环，不停地从服务器端读取数据。

作者：昨夜星辰