Source：資料來源、獲取資料資料傳遞功能

Sink：消耗資料，傳送資料

Source->Sink：Source獲取網路資料，Sink消耗資料寫入檔案

注意：live555兩個迴圈一個是傳送RTP資料，一個就是讀取RTP即   SingleStep();

1. void BasicTaskScheduler0::doEventLoop(char* watchVariable)   
2. {  
3.   // Repeatedly loop, handling readble sockets and timed events:
4.   while (1)   
5.   {  
6.     if (watchVariable != NULL && *watchVariable != 0) 
   break;  
7.     SingleStep();  
8.   }  
9. }  

從這裡可知，live555在客戶端處理資料實際上是單執行緒的程式，不斷執行SingleStep()函式中的程式碼。通過檢視該函式程式碼裡，下面一句程式碼為重點

1. (*handler->handlerProc)(handler->clientData, resultConditionSet);  

其中該條程式碼出現了兩次，通過除錯跟蹤它的執行軌跡，第一次出現呼叫的函式是為了處理和RTSP伺服器的通訊協議的商定，而第二次出現呼叫的函式才是處理真正的視訊和音訊資料。對於RTSP通訊協議的分析我們暫且不討論，而直接進入第二次呼叫該函式的部分。

在我們的除錯過程中在執行到上面的函式時就直接呼叫到livemedia目錄下的如下函式

1. void MultiFramedRTPSource::networkReadHandler(MultiFramedRTPSource* source, int/*mask*/)   
2. {  
3.   source->networkReadHandler1();  
4. }  

//下面這個函式實現的主要功能就是從socket端讀取資料並存儲資料

1. void MultiFramedRTPSource::networkReadHandler1()   
2. {  
3.   BufferedPacket* bPacket = fPacketReadInProgress;  
4.   if (bPacket == NULL)  
5.   {  
6.     // Normal case: Get a free BufferedPacket descriptor to hold the new network packet:
7.     //分配一塊新的儲存空間來儲存從socket端讀取的資料
8.     bPacket = fReorderingBuffer->getFreePacket(this);  
9.   }  
10.   // Read the network packet, and perform sanity checks on the RTP header:
11.   Boolean readSuccess = False;  
12.   do
13.   {  
14.     Boolean packetReadWasIncomplete = fPacketReadInProgress != NULL;  
15.     //fillInData()函式封裝了從socket端獲取資料的過程，到此函式執行完已經將資料儲存到了bPacket物件中
16.     if (!bPacket->fillInData(fRTPInterface, packetReadWasIncomplete))   
17.    {  
18.       if (bPacket->bytesAvailable() == 0)   
19.       {  
20.       envir() << "MultiFramedRTPSource error: Hit limit when reading incoming packet over TCP. Increase \"MAX_PACKET_SIZE\"\n";  
21.       }  
22.       break;  
23.    }  
24.     if (packetReadWasIncomplete)  
25.     {  
26.       // We need additional read(s) before we can process the incoming packet:
27.       fPacketReadInProgress = bPacket;  
28.       return;  
29.     } else
30.     {  
31.       fPacketReadInProgress = NULL;  
32.     }  
33.     //省略關於RTP包的處理
34.     ...  
35.     ...  
36.     ...  
37.     //fReorderingBuffer為MultiFramedRTPSource類中的物件，該物件建立了一個儲存Packet資料包物件的連結串列
38.     //下面的storePacket()函式即將上面獲取的資料包儲存在連結串列中
39.     if (!fReorderingBuffer->storePacket(bPacket)) break;   
40.     readSuccess = True;  
41.   } while (0);  
42.   if (!readSuccess) fReorderingBuffer->freePacket(bPacket);  
43.   doGetNextFrame1();  
44.   // If we didn't get proper data this time, we'll get another chance
45. }  

//下面的這個函式則實現從上面函式中介紹的儲存資料包連結串列的物件（即fReorderingBuffer）中取出資料包並呼叫相應函式使用它

//程式碼1.1

1. void MultiFramedRTPSource::doGetNextFrame1()   
2. {  
3.   while (fNeedDelivery)   
4.   {  
5.     // If we already have packet data available, then deliver it now.
6.     Boolean packetLossPrecededThis;   
7.     //從fReorderingBuffer物件中取出一個數據包
8.     BufferedPacket* nextPacket  
9.       = fReorderingBuffer->getNextCompletedPacket(packetLossPrecededThis);  
10.     if (nextPacket == NULL) break;  
11.     fNeedDelivery = False;  
12.     if (nextPacket->useCount() == 0)   
13.     {  
14.       // Before using the packet, check whether it has a special header
15.       // that needs to be processed:
16.       unsigned specialHeaderSize;  
17.       if (!processSpecialHeader(nextPacket, specialHeaderSize))  
18.       {  
19.         // Something's wrong with the header; reject the packet:
20.         fReorderingBuffer->releaseUsedPacket(nextPacket);  
21.         fNeedDelivery = True;  
22.         break;  
23.       }  
24.       nextPacket->skip(specialHeaderSize);  
25.     }  
26.     // Check whether we're part of a multi-packet frame, and whether
27.     // there was packet loss that would render this packet unusable:
28.     if (fCurrentPacketBeginsFrame)   
29.     {  
30.       if (packetLossPrecededThis || fPacketLossInFragmentedFrame)   
31.       {  
32.         // We didn't get all of the previous frame.
33.         // Forget any data that we used from it:
34.         fTo = fSavedTo; fMaxSize = fSavedMaxSize;  
35.         fFrameSize = 0;  
36.       }  
37.       fPacketLossInFragmentedFrame = False;  
38.     } elseif (packetLossPrecededThis)   
39.     {  
40.       // We're in a multi-packet frame, with preceding packet loss
41.       fPacketLossInFragmentedFrame = True;  
42.     }  
43.     if (fPacketLossInFragmentedFrame)  
44.     {  
45.       // This packet is unusable; reject it:
46.       fReorderingBuffer->releaseUsedPacket(nextPacket);  
47.       fNeedDelivery = True;  
48.       break;  
49.     }  
50.     // The packet is usable. Deliver all or part of it to our caller:
51.     unsigned frameSize;  
52.     //將上面取出的資料包拷貝到fTo指標所指向的地址
53.     nextPacket->use(fTo, fMaxSize, frameSize, fNumTruncatedBytes,  
54.             fCurPacketRTPSeqNum, fCurPacketRTPTimestamp,  
55.             fPresentationTime, fCurPacketHasBeenSynchronizedUsingRTCP,  
56.             fCurPacketMarkerBit);  
57.     fFrameSize += frameSize;  
58.     if (!nextPacket->hasUsableData())   
59.     {  
60.       // We're completely done with this packet now
61.       fReorderingBuffer->releaseUsedPacket(nextPacket);  
62.     }  
63.     if (fCurrentPacketCompletesFrame) //如果完整的取出了一幀資料，則可呼叫需要該幀資料的函式去處理它
64.      {  
65.       // We have all the data that the client wants.
66.       if (fNumTruncatedBytes > 0)   
67.       {  
68.     envir() << "MultiFramedRTPSource::doGetNextFrame1(): The total received frame size exceeds the client's buffer size ("
69.         << fSavedMaxSize << ").  "
70.         << fNumTruncatedBytes << " bytes of trailing data will be dropped!\n";  
71.       }  
72.       // Call our own 'after getting' function, so that the downstream object can consume the data:
73.       if (fReorderingBuffer->isEmpty())   
74.       {  
75.         // Common case optimization: There are no more queued incoming packets, so this code will not get
76.         // executed again without having first returned to the event loop.  Call our 'after getting' function
77.         // directly, because there's no risk of a long chain of recursion (and thus stack overflow):
78.     afterGetting(this);  //呼叫函式去處理取出的資料幀
79.        } else
80.       {  
81.     // Special case: Call our 'after getting' function via the event loop.
82.     nextTask() = envir().taskScheduler().scheduleDelayedTask(0,  
83.                                  (TaskFunc*)FramedSource::afterGetting, this);  
84.       }  
85.     }  
86.     else
87.     {  
88.       // This packet contained fragmented data, and does not complete
89.       // the data that the client wants.  Keep getting data:
90.       fTo += frameSize; fMaxSize -= frameSize;  
91.       fNeedDelivery = True;  
92.     }  
93.   }  
94. }  

//下面這個函式即開始呼叫執行需要該幀資料的函式

1. void FramedSource::afterGetting(FramedSource* source)   
2. {  
3.   source->fIsCurrentlyAwaitingData = False;  
4.       // indicates that we can be read again
5.       // Note that this needs to be done here, in case the "fAfterFunc"
6.       // called below tries to read another frame (which it usually will)
7.   if (source->fAfterGettingFunc != NULL)     

1.   {  
2.     (*(source->fAfterGettingFunc))(source->fAfterGettingClientData,  
3.                    source->fFrameSize, source->fNumTruncatedBytes,  
4.                    source->fPresentationTime,  
5.                    source->fDurationInMicroseconds);  
6.   }  
7. }  

上面的fAfterGettingFunc為我們自己註冊的函式，如果執行的是testProgs中的openRTSP例項，則該函式指向下列程式碼中通過呼叫getNextFrame（）註冊的afterGettingFrame（）函式

1. Boolean FileSink::continuePlaying()  
2. {  
3.   if (fSource == NULL) return False;  
4.   fSource->getNextFrame(fBuffer, fBufferSize,  
5.             afterGettingFrame, this,  
6.             onSourceClosure, this);  
7.   return True;  
8. }  

如果執行的是testProgs中的testRTSPClient中的例項，則該函式指向這裡註冊的afterGettingFrame（）函式

1. Boolean DummySink::continuePlaying()  
2. {  
3.   if (fSource == NULL) return False; // sanity check (should not happen)
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. }  

從上面的程式碼中可以看到getNextFrame()函式的第一個引數為分別在各自類中定義的buffer，我們繼續以openRTSP為執行程式來分析，fBuffer為FileSink類裡定義的指標：unsigned char* fBuffer;

這裡我們先繞一個彎，看看getNextFrame()函式裡做了什麼

1. void FramedSource::getNextFrame(unsigned char* to, unsigned maxSize,  
2.                 afterGettingFunc* afterGettingFunc,  
3.                 void* afterGettingClientData,  
4.                 onCloseFunc* onCloseFunc,  
5.                 void* onCloseClientData)   
6. {  
7.   // Make sure we're not already being read:
8.   if (fIsCurrentlyAwaitingData)     
9.   {  
10.     envir() << "FramedSource[" << this << "]::getNextFrame(): attempting to read more than once at the same time!\n";  
11.     envir().internalError();  
12.   }  
13.   fTo = to;  
14.   fMaxSize = maxSize;  
15.   fNumTruncatedBytes = 0; // by default; could be changed by doGetNextFrame()
16.   fDurationInMicroseconds = 0; // by default; could be changed by doGetNextFrame()
17.   fAfterGettingFunc = afterGettingFunc;  
18.   fAfterGettingClientData = afterGettingClientData;  
19.   fOnCloseFunc = onCloseFunc;  
20.   fOnCloseClientData = onCloseClientData;  
21.   fIsCurrentlyAwaitingData = True;  
22.   doGetNextFrame();  
23. }  

從程式碼可以知道上面getNextFrame()中傳入的第一個引數fBuffer指向了指標fTo，而我們在前面分析程式碼1.1中的void MultiFramedRTPSource::doGetNextFrame1()函式中有下面一段程式碼：

1. //將上面取出的資料包拷貝到fTo指標所指向的地址
2.  nextPacket->use(fTo, fMaxSize, frameSize, fNumTruncatedBytes,  
3.    fCurPacketRTPSeqNum, fCurPacketRTPTimestamp,  
4.    fPresentationTime, fCurPacketHasBeenSynchronizedUsingRTCP,  
5.    fCurPacketMarkerBit);  

實際上現在應該明白了，從getNextFrame()函式中傳入的第一個引數fBuffer最終儲存的即是從資料包連結串列物件中取出的資料，並且在呼叫上面的use()函式後就可以使用了。
而在void MultiFramedRTPSource::doGetNextFrame1()函式中程式碼顯示的最終呼叫我們註冊的void FileSink::afterGettingFrame（）正好是在use()函式呼叫之後的afterGetting(this)中呼叫。我們再看看afterGettingFrame()做了什麼處理：

1. void FileSink::afterGettingFrame(void* clientData, unsigned frameSize,  
2.                  unsigned numTruncatedBytes,  
3.                  struct timeval presentationTime,  
4.                  unsigned /*durationInMicroseconds*/)  
5. {  
6.   FileSink* sink = (FileSink*)clientData;  
7.   sink->afterGettingFrame(frameSize, numTruncatedBytes, presentationTime);  
8. }  
9. void FileSink::afterGettingFrame(unsigned frameSize,  
10.                  unsigned numTruncatedBytes,  
11.                  struct timeval presentationTime)   
12. {  
13.   if (numTruncatedBytes > 0)     
14.   {  
15.     envir() << "FileSink::afterGettingFrame(): The input frame data was too large for our buffer size ("
16.         << fBufferSize << ").  "
17.             << numTruncatedBytes << " bytes of trailing data was dropped!  Correct this by increasing the \"bufferSize\" parameter in the \"createNew()\" call to at least "
18.             << fBufferSize + numTruncatedBytes << "\n";  
19.   }  
20.   addData(fBuffer, frameSize, presentationTime);  
21.   if (fOutFid == NULL || fflush(fOutFid) == EOF)     
22.   {  
23.     // The output file has closed.  Handle this the same way as if the
24.     // input source had closed:
25.     onSourceClosure(this);  
26.     stopPlaying();  
27.     return;  
28.   }  
29.   if (fPerFrameFileNameBuffer != NULL)     
30.   {  
31.     if (fOutFid != NULL) { fclose(fOutFid); fOutFid = NULL; }  
32.   }  
33.   // Then try getting the next frame:
34.   continuePlaying();  
35. }  

從上面程式碼可以看到呼叫了addData()函式將資料儲存到檔案中，然後繼續continuePlaying()又去獲取下一幀資料然後處理，直到遇到迴圈結束然後依次退出呼叫函式。最後看看addData()函式的實現即可知：

1. void FileSink::addData(unsigned charconst* data, unsigned dataSize,  
2.                struct timeval presentationTime)   
3. {  
4.   if (fPerFrameFileNameBuffer != NULL)     
5.   {  
6.     // Special case: Open a new file on-the-fly for this frame
7.     sprintf(fPerFrameFileNameBuffer, "%s-%lu.%06lu", fPerFrameFileNamePrefix,  
8.         presentationTime.tv_sec, presentationTime.tv_usec);  
9.     fOutFid = OpenOutputFile(envir(), fPerFrameFileNameBuffer);  
10.   }  
11.   // Write to our file:
12. #ifdef TEST_LOSS
13.   static unsigned const framesPerPacket = 10;  
14.   static unsigned const frameCount = 0;  
15.   static Boolean const packetIsLost;  
16.   if ((frameCount++)%framesPerPacket == 0)     
17.   {  
18.     packetIsLost = (our_random()%10 == 0); // simulate 10% packet loss #####
19.   }  
20.   if (!packetIsLost)  
21. #endif
22.   if (fOutFid != NULL && data != NULL)    
23.   {  
24.     fwrite(data, 1, dataSize, fOutFid);  
25.   }  
26. }  

最後呼叫系統函式fwrite()實現寫入檔案功能。

總結：從上面的分析可知，如果要取得從RTSP伺服器端接收並儲存的資料幀，我們只需要定義一個類並實現如下格式兩個的函式，並宣告一個指標地址buffer用於指向資料幀，再在continuePlaying()函式中呼叫getNextFrame(buffer，...)即可。

1. typedefvoid (afterGettingFunc)(void* clientData, unsigned frameSize,  
2.           unsigned numTruncatedBytes,  
3.           struct timeval presentationTime,  
4.           unsigned durationInMicroseconds);  
5. typedefvoid (onCloseFunc)(void* clientData);  

然後再在afterGettingFunc的函式中即可使用buffer。.