Live555学习之(四)------建立RTSP连接的过程（RTSP客户端）

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

　　首先来看一下main函数：

char eventLoopWatchVariable = 0;

int main(int argc, char** argv) {
  // Begin by setting up our usage environment:
  TaskScheduler* scheduler = BasicTaskScheduler::createNew();
  UsageEnvironment* env = BasicUsageEnvironment::createNew(*scheduler);

  // We need at least one "rtsp://" URL argument:
  if (argc < 2) {
    usage(*env, argv[0]);
    return 1;
  }

  // There are argc-1 URLs: argv[1] through argv[argc-1].  Open and start streaming each one:
  for (int i = 1; i <= argc-1; ++i) {
    openURL(*env, argv[0], argv[i]);
  }

  // All subsequent activity takes place within the event loop:
  env->taskScheduler().doEventLoop(&eventLoopWatchVariable);
    // This function call does not return, unless, at some point in time, "eventLoopWatchVariable" gets set to something non-zero.

  return 0;

  // If you choose to continue the application past this point (i.e., if you comment out the "return 0;" statement above),
  // and if you don't intend to do anything more with the "TaskScheduler" and "UsageEnvironment" objects,
  // then you can also reclaim the (small) memory used by these objects by uncommenting the following code:
  /*
    env->reclaim(); env = NULL;
    delete scheduler; scheduler = NULL;
  */
}

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

void openURL(UsageEnvironment& env, char const* progName, char const* rtspURL) {
  // Begin by creating a "RTSPClient" object.  Note that there is a separate "RTSPClient" object for each stream that we wish
  // to receive (even if more than stream uses the same "rtsp://" URL).
  RTSPClient* rtspClient = ourRTSPClient::createNew(env, rtspURL, RTSP_CLIENT_VERBOSITY_LEVEL, progName);
  if (rtspClient == NULL) {
    env << "Failed to create a RTSP client for URL "" << rtspURL << "": " << env.getResultMsg() << "
";
    return;
  }

  ++rtspClientCount;

  // Next, send a RTSP "DESCRIBE" command, to get a SDP description for the stream.
  // Note that this command - like all RTSP commands - is sent asynchronously; we do not block, waiting for a response.
  // Instead, the following function call returns immediately, and we handle the RTSP response later, from within the event loop:
  rtspClient->sendDescribeCommand(continueAfterDESCRIBE); //发送DESCRIBE命令，并传入回调函数
}

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

void continueAfterDESCRIBE(RTSPClient* rtspClient, int resultCode, char* resultString) {
  do {
    UsageEnvironment& env = rtspClient->envir(); // alias
    StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias

    if (resultCode != 0) {  // 返回结果码非0表示出错
      env << *rtspClient << "Failed to get a SDP description: " << resultString << "
";
      delete[] resultString;
      break;
    }
    // resultString即从服务器端返回的SDP信息字符串
    char* const sdpDescription = resultString;
    env << *rtspClient << "Got a SDP description:
" << sdpDescription << "
";

    // Create a media session object from this SDP description:
    scs.session = MediaSession::createNew(env, sdpDescription);   //根据SDP信息创建一个MediaSession对象
    delete[] sdpDescription; // because we don't need it anymore
    if (scs.session == NULL) {
      env << *rtspClient << "Failed to create a MediaSession object from the SDP description: " << env.getResultMsg() << "
";
      break;
    } else if (!scs.session->hasSubsessions()) {
      env << *rtspClient << "This session has no media subsessions (i.e., no "m=" lines)
";
      break;
    }

    // Then, create and set up our data source objects for the session.  We do this by iterating over the session's 'subsessions',
    // calling "MediaSubsession::initiate()", and then sending a RTSP "SETUP" command, on each one.
    // (Each 'subsession' will have its own data source.)
    scs.iter = new MediaSubsessionIterator(*scs.session);
    setupNextSubsession(rtspClient);　　　　　　　　//开始对服务器端的每个ServerMediaSubsession发送SETUP命令请求建立连接
    return;
  } while (0);

  // An unrecoverable error occurred with this stream.
  shutdownStream(rtspClient);
}

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

void setupNextSubsession(RTSPClient* rtspClient) {
  UsageEnvironment& env = rtspClient->envir(); // alias
  StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias
  
  scs.subsession = scs.iter->next();
  if (scs.subsession != NULL) {
    if (!scs.subsession->initiate()) {  // 调用initiate函数初始化MediaSubsession对象
      env << *rtspClient << "Failed to initiate the "" << *scs.subsession << "" subsession: " << env.getResultMsg() << "
";
      setupNextSubsession(rtspClient); // give up on this subsession; go to the next one
    } else {
      env << *rtspClient << "Initiated the "" << *scs.subsession << "" subsession (";
      if (scs.subsession->rtcpIsMuxed()) {
    env << "client port " << scs.subsession->clientPortNum();
      } else {
    env << "client ports " << scs.subsession->clientPortNum() << "-" << scs.subsession->clientPortNum()+1;
      }
      env << ")
";
      // 发送SETUP命令
      // Continue setting up this subsession, by sending a RTSP "SETUP" command:
      rtspClient->sendSetupCommand(*scs.subsession, continueAfterSETUP, False, REQUEST_STREAMING_OVER_TCP);
    }
    return;
  }
  // 成功与所有的ServerMediaSubsession建立了连接，现在发送PLAY命令
  // We've finished setting up all of the subsessions.  Now, send a RTSP "PLAY" command to start the streaming:
  if (scs.session->absStartTime() != NULL) {
    // Special case: The stream is indexed by 'absolute' time, so send an appropriate "PLAY" command:
    rtspClient->sendPlayCommand(*scs.session, continueAfterPLAY, scs.session->absStartTime(), scs.session->absEndTime());
  } else {
    scs.duration = scs.session->playEndTime() - scs.session->playStartTime();
    rtspClient->sendPlayCommand(*scs.session, continueAfterPLAY);
  }
}

void continueAfterSETUP(RTSPClient* rtspClient, int resultCode, char* resultString) {
  do {
    UsageEnvironment& env = rtspClient->envir(); // alias
    StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias

    if (resultCode != 0) {
      env << *rtspClient << "Failed to set up the "" << *scs.subsession << "" subsession: " << resultString << "
";
      break;
    }

    env << *rtspClient << "Set up the "" << *scs.subsession << "" subsession (";
    if (scs.subsession->rtcpIsMuxed()) {
      env << "client port " << scs.subsession->clientPortNum();
    } else {
      env << "client ports " << scs.subsession->clientPortNum() << "-" << scs.subsession->clientPortNum()+1;
    }
    env << ")
";

    // Having successfully setup the subsession, create a data sink for it, and call "startPlaying()" on it.
    // (This will prepare the data sink to receive data; the actual flow of data from the client won't start happening until later,
    // after we've sent a RTSP "PLAY" command.)
    //对每个MediaSubsession创建一个MediaSink对象来请求和保存数据
    scs.subsession->sink = DummySink::createNew(env, *scs.subsession, rtspClient->url());
      // perhaps use your own custom "MediaSink" subclass instead
    if (scs.subsession->sink == NULL) {
      env << *rtspClient << "Failed to create a data sink for the "" << *scs.subsession
      << "" subsession: " << env.getResultMsg() << "
";
      break;
    }

    env << *rtspClient << "Created a data sink for the "" << *scs.subsession << "" subsession
";
    scs.subsession->miscPtr = rtspClient; // a hack to let subsession handle functions get the "RTSPClient" from the subsession 
    scs.subsession->sink->startPlaying(*(scs.subsession->readSource()),
                       subsessionAfterPlaying, scs.subsession);          // 调用MediaSink的startPlaying函数准备播放
    // Also set a handler to be called if a RTCP "BYE" arrives for this subsession:
    if (scs.subsession->rtcpInstance() != NULL) {
      scs.subsession->rtcpInstance()->setByeHandler(subsessionByeHandler, scs.subsession);
    }
  } while (0);
  delete[] resultString;

  // Set up the next subsession, if any:  与下一个ServerMediaSubsession建立连接
  setupNextSubsession(rtspClient);
}

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

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

Boolean MediaSubsession::initiate(int useSpecialRTPoffset) {
  if (fReadSource != NULL) return True; // has already been initiated

  do {
    if (fCodecName == NULL) {
      env().setResultMsg("Codec is unspecified");
      break;
    }
    //创建客户端socket，包括RTP socket和RTCP socket，准备从服务器端接收数据
    // Create RTP and RTCP 'Groupsocks' on which to receive incoming data.
    // (Groupsocks will work even for unicast addresses)
    struct in_addr tempAddr;
    tempAddr.s_addr = connectionEndpointAddress();
        // This could get changed later, as a result of a RTSP "SETUP"
    //使用指定的RTP端口和RTCP端口，RTP端口必须是偶数，而RTCP端口必须是（RTP端口+1）
    if (fClientPortNum != 0 && (honorSDPPortChoice || IsMulticastAddress(tempAddr.s_addr))) {
      // The sockets' port numbers were specified for us.  Use these:
      Boolean const protocolIsRTP = strcmp(fProtocolName, "RTP") == 0;
      if (protocolIsRTP && !fMultiplexRTCPWithRTP) {
    fClientPortNum = fClientPortNum&~1;
        // use an even-numbered port for RTP, and the next (odd-numbered) port for RTCP
      }
      if (isSSM()) {
    fRTPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, fClientPortNum);
      } else {
    fRTPSocket = new Groupsock(env(), tempAddr, fClientPortNum, 255);
      }
      if (fRTPSocket == NULL) {
    env().setResultMsg("Failed to create RTP socket");
    break;
      }
      
      if (protocolIsRTP) {
    if (fMultiplexRTCPWithRTP) {
      // Use the RTP 'groupsock' object for RTCP as well:
      fRTCPSocket = fRTPSocket;
    } else {
      // Set our RTCP port to be the RTP port + 1:
      portNumBits const rtcpPortNum = fClientPortNum|1;
      if (isSSM()) {
        fRTCPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, rtcpPortNum);
      } else {
        fRTCPSocket = new Groupsock(env(), tempAddr, rtcpPortNum, 255);
      }
    }
      }
    } else {
          // 选取随机的RTP端口和RTCP端口
      // Port numbers were not specified in advance, so we use ephemeral port numbers.
      // Create sockets until we get a port-number pair (even: RTP; even+1: RTCP).
      // (However, if we're multiplexing RTCP with RTP, then we create only one socket,
      // and the port number  can be even or odd.)
      // We need to make sure that we don't keep trying to use the same bad port numbers over
      // and over again, so we store bad sockets in a table, and delete them all when we're done.
      HashTable* socketHashTable = HashTable::create(ONE_WORD_HASH_KEYS);
      if (socketHashTable == NULL) break;
      Boolean success = False;
      NoReuse dummy(env());
          // ensures that our new ephemeral port number won't be one that's already in use

      while (1) {
    // Create a new socket:
    if (isSSM()) {
      fRTPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, 0);
    } else {
      fRTPSocket = new Groupsock(env(), tempAddr, 0, 255);
    }
    if (fRTPSocket == NULL) {
      env().setResultMsg("MediaSession::initiate(): unable to create RTP and RTCP sockets");
      break;
    }

    // Get the client port number:
    Port clientPort(0);
    if (!getSourcePort(env(), fRTPSocket->socketNum(), clientPort)) {
      break;
    }
    fClientPortNum = ntohs(clientPort.num()); 

    if (fMultiplexRTCPWithRTP) {
      // Use this RTP 'groupsock' object for RTCP as well:
      fRTCPSocket = fRTPSocket;
      success = True;
      break;
    }      

    // To be usable for RTP, the client port number must be even:
    if ((fClientPortNum&1) != 0) { // it's odd
      // Record this socket in our table, and keep trying:
      unsigned key = (unsigned)fClientPortNum;
      Groupsock* existing = (Groupsock*)socketHashTable->Add((char const*)key, fRTPSocket);
      delete existing; // in case it wasn't NULL
      continue;
    }

    // Make sure we can use the next (i.e., odd) port number, for RTCP:
    portNumBits rtcpPortNum = fClientPortNum|1;
    if (isSSM()) {
      fRTCPSocket = new Groupsock(env(), tempAddr, fSourceFilterAddr, rtcpPortNum);
    } else {
      fRTCPSocket = new Groupsock(env(), tempAddr, rtcpPortNum, 255);
    }
    if (fRTCPSocket != NULL && fRTCPSocket->socketNum() >= 0) {
      // Success! Use these two sockets.
      success = True;
      break;
    } else {
      // We couldn't create the RTCP socket (perhaps that port number's already in use elsewhere?).
      delete fRTCPSocket; fRTCPSocket = NULL;

      // Record the first socket in our table, and keep trying:
      unsigned key = (unsigned)fClientPortNum;
      Groupsock* existing = (Groupsock*)socketHashTable->Add((char const*)key, fRTPSocket);
      delete existing; // in case it wasn't NULL
      continue;
    }
      }

      // Clean up the socket hash table (and contents):
      Groupsock* oldGS;
      while ((oldGS = (Groupsock*)socketHashTable->RemoveNext()) != NULL) {
    delete oldGS;
      }
      delete socketHashTable;

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

    // Try to use a big receive buffer for RTP - at least 0.1 second of
    // specified bandwidth and at least 50 KB
    unsigned rtpBufSize = fBandwidth * 25 / 2; // 1 kbps * 0.1 s = 12.5 bytes
    if (rtpBufSize < 50 * 1024)
      rtpBufSize = 50 * 1024;
    increaseReceiveBufferTo(env(), fRTPSocket->socketNum(), rtpBufSize);

    if (isSSM() && fRTCPSocket != NULL) {
      // Special case for RTCP SSM: Send RTCP packets back to the source via unicast:
      fRTCPSocket->changeDestinationParameters(fSourceFilterAddr,0,~0);
    }
    //创建FramedSource对象来请求数据
    // Create "fRTPSource" and "fReadSource":
    if (!createSourceObjects(useSpecialRTPoffset)) break;

    if (fReadSource == NULL) {
      env().setResultMsg("Failed to create read source");
      break;
    }
    // 创建RTCPInstance对象
    // Finally, create our RTCP instance. (It starts running automatically)
    if (fRTPSource != NULL && fRTCPSocket != NULL) {
      // If bandwidth is specified, use it and add 5% for RTCP overhead.
      // Otherwise make a guess at 500 kbps.
      unsigned totSessionBandwidth
    = fBandwidth ? fBandwidth + fBandwidth / 20 : 500;
      fRTCPInstance = RTCPInstance::createNew(env(), fRTCPSocket,
                          totSessionBandwidth,
                          (unsigned char const*)
                          fParent.CNAME(),
                          NULL /* we're a client */,
                          fRTPSource);
      if (fRTCPInstance == NULL) {
    env().setResultMsg("Failed to create RTCP instance");
    break;
      }
    }

    return True;
  } while (0);

  deInitiate();
  fClientPortNum = 0;
  return False;
}

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

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

void continueAfterPLAY(RTSPClient* rtspClient, int resultCode, char* resultString) {
  Boolean success = False;

  do {
    UsageEnvironment& env = rtspClient->envir(); // alias
    StreamClientState& scs = ((ourRTSPClient*)rtspClient)->scs; // alias

    if (resultCode != 0) {
      env << *rtspClient << "Failed to start playing session: " << resultString << "
";
      break;
    }

    // Set a timer to be handled at the end of the stream's expected duration (if the stream does not already signal its end
    // using a RTCP "BYE").  This is optional.  If, instead, you want to keep the stream active - e.g., so you can later
    // 'seek' back within it and do another RTSP "PLAY" - then you can omit this code.
    // (Alternatively, if you don't want to receive the entire stream, you could set this timer for some shorter value.)
    if (scs.duration > 0) {
      unsigned const delaySlop = 2; // number of seconds extra to delay, after the stream's expected duration.  (This is optional.)
      scs.duration += delaySlop;
      unsigned uSecsToDelay = (unsigned)(scs.duration*1000000);
      scs.streamTimerTask = env.taskScheduler().scheduleDelayedTask(uSecsToDelay, (TaskFunc*)streamTimerHandler, rtspClient);
    }

    env << *rtspClient << "Started playing session";
    if (scs.duration > 0) {
      env << " (for up to " << scs.duration << " seconds)";
    }
    env << "...
";

    success = True;
  } while (0);
  delete[] resultString;

  if (!success) {
    // An unrecoverable error occurred with this stream.
    shutdownStream(rtspClient);
  }
}

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

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

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

Boolean MediaSink::startPlaying(MediaSource& source,
                afterPlayingFunc* afterFunc,
                void* afterClientData) {
  // Make sure we're not already being played:
  if (fSource != NULL) {
    envir().setResultMsg("This sink is already being played");
    return False;
  }

  // Make sure our source is compatible:
  if (!sourceIsCompatibleWithUs(source)) {
    envir().setResultMsg("MediaSink::startPlaying(): source is not compatible!");
    return False;
  }
  fSource = (FramedSource*)&source;       //此处的fSource是之前创立的H264VideoRTPSource对象

  fAfterFunc = afterFunc;
  fAfterClientData = afterClientData;
  return continuePlaying();
}

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

Boolean DummySink::continuePlaying() {
  if (fSource == NULL) return False; // sanity check (should not happen)

  // Request the next frame of data from our input source.  "afterGettingFrame()" will get called later, when it arrives:
  fSource->getNextFrame(fReceiveBuffer, DUMMY_SINK_RECEIVE_BUFFER_SIZE,
                        afterGettingFrame, this,
                        onSourceClosure, this);
  return True;
}

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

void MultiFramedRTPSource::doGetNextFrame() {
  if (!fAreDoingNetworkReads) {
    // Turn on background read handling of incoming packets:
    fAreDoingNetworkReads = True;
    TaskScheduler::BackgroundHandlerProc* handler
      = (TaskScheduler::BackgroundHandlerProc*)&networkReadHandler;
    fRTPInterface.startNetworkReading(handler);　　//通过RTPInterface对象读取网络数据，在服务器端是通过RTPInterface对象发送网络数据
　　　　//读到数据后回调networkReadHandler函数来处理
  }

  fSavedTo = fTo;　　　　　　　　　　　　//读到的数据保存在fTo中
  fSavedMaxSize = fMaxSize;
  fFrameSize = 0; // for now
  fNeedDelivery = True;
  doGetNextFrame1();
}

void MultiFramedRTPSource::doGetNextFrame1() {
  while (fNeedDelivery) {　　　　　　　　　　　　　　　　　　　　　　　　　　19     // If we already have packet data available, then deliver it now.
    Boolean packetLossPrecededThis;
    BufferedPacket* nextPacket
      = fReorderingBuffer->getNextCompletedPacket(packetLossPrecededThis);
    if (nextPacket == NULL) break;

    fNeedDelivery = False;

    if (nextPacket->useCount() == 0) {
      // Before using the packet, check whether it has a special header
      // that needs to be processed:
      unsigned specialHeaderSize;
      if (!processSpecialHeader(nextPacket, specialHeaderSize)) {
    // Something's wrong with the header; reject the packet:
    fReorderingBuffer->releaseUsedPacket(nextPacket);
    fNeedDelivery = True;
    break;
      }
      nextPacket->skip(specialHeaderSize);
    }

    // Check whether we're part of a multi-packet frame, and whether
    // there was packet loss that would render this packet unusable:
    if (fCurrentPacketBeginsFrame) {
      if (packetLossPrecededThis || fPacketLossInFragmentedFrame) {
    // We didn't get all of the previous frame.
    // Forget any data that we used from it:
    fTo = fSavedTo; fMaxSize = fSavedMaxSize;
    fFrameSize = 0;
      }
      fPacketLossInFragmentedFrame = False;
    } else if (packetLossPrecededThis) {
      // We're in a multi-packet frame, with preceding packet loss
      fPacketLossInFragmentedFrame = True;
    }
    if (fPacketLossInFragmentedFrame) {
      // This packet is unusable; reject it:
      fReorderingBuffer->releaseUsedPacket(nextPacket);
      fNeedDelivery = True;
      break;
    }

    // The packet is usable. Deliver all or part of it to our caller:
    unsigned frameSize;
    nextPacket->use(fTo, fMaxSize, frameSize, fNumTruncatedBytes,
            fCurPacketRTPSeqNum, fCurPacketRTPTimestamp,
            fPresentationTime, fCurPacketHasBeenSynchronizedUsingRTCP,
            fCurPacketMarkerBit);
    fFrameSize += frameSize;

    if (!nextPacket->hasUsableData()) {
      // We're completely done with this packet now
      fReorderingBuffer->releaseUsedPacket(nextPacket);
    }

    if (fCurrentPacketCompletesFrame) {         // 成功读到一帧数据
      // We have all the data that the client wants.
      if (fNumTruncatedBytes > 0) {
    envir() << "MultiFramedRTPSource::doGetNextFrame1(): The total received frame size exceeds the client's buffer size ("
        << fSavedMaxSize << ").  "
        << fNumTruncatedBytes << " bytes of trailing data will be dropped!
";
      }
      // Call our own 'after getting' function, so that the downstream object can consume the data:
      if (fReorderingBuffer->isEmpty()) {
    // Common case optimization: There are no more queued incoming packets, so this code will not get
    // executed again without having first returned to the event loop.  Call our 'after getting' function
    // directly, because there's no risk of a long chain of recursion (and thus stack overflow):
    afterGetting(this);
      } else {
    // Special case: Call our 'after getting' function via the event loop.
    nextTask() = envir().taskScheduler().scheduleDelayedTask(0,
                                 (TaskFunc*)FramedSource::afterGetting, this);
      }
    } else {
      // This packet contained fragmented data, and does not complete
      // the data that the client wants.  Keep getting data:
      fTo += frameSize; fMaxSize -= frameSize;
      fNeedDelivery = True;
    }
  }
}

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

void DummySink::afterGettingFrame(void* clientData, unsigned frameSize, unsigned numTruncatedBytes,
                  struct timeval presentationTime, unsigned durationInMicroseconds) {
  DummySink* sink = (DummySink*)clientData;
  sink->afterGettingFrame(frameSize, numTruncatedBytes, presentationTime, durationInMicroseconds);
}

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

void DummySink::afterGettingFrame(unsigned frameSize, unsigned numTruncatedBytes,
                  struct timeval presentationTime, unsigned /*durationInMicroseconds*/) {
  // We've just received a frame of data.  (Optionally) print out information about it:
#ifdef DEBUG_PRINT_EACH_RECEIVED_FRAME
  if (fStreamId != NULL) envir() << "Stream "" << fStreamId << ""; ";
  envir() << fSubsession.mediumName() << "/" << fSubsession.codecName() << ":	Received " << frameSize << " bytes";
  if (numTruncatedBytes > 0) envir() << " (with " << numTruncatedBytes << " bytes truncated)";
  char uSecsStr[6+1]; // used to output the 'microseconds' part of the presentation time
  sprintf(uSecsStr, "%06u", (unsigned)presentationTime.tv_usec);
  envir() << ".	Presentation time: " << (int)presentationTime.tv_sec << "." << uSecsStr;
  if (fSubsession.rtpSource() != NULL && !fSubsession.rtpSource()->hasBeenSynchronizedUsingRTCP()) {
    envir() << "!"; // mark the debugging output to indicate that this presentation time is not RTCP-synchronized
  }
#ifdef DEBUG_PRINT_NPT
  envir() << "	NPT: " << fSubsession.getNormalPlayTime(presentationTime);
#endif
  envir() << "
";
#endif
  
  // Then continue, to request the next frame of data:
  continuePlaying();
}

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