live555源码分析

    live555支持的文件格式多为单流的文件，仅支持*.mpg、*.mkv、*.webm几种音视频混合类型的文件。其实我的目的是扩展其支持的格式，如avi等， 所以来分析一下mpg文件的处理。


    mpg文件处理涉及的类相当多，类间的关系比较复杂，对于RTP打包的过程(在RTPSink中完成)所以有的媒体类型比较相似(细节上有差异，这些都是在特定媒体相关的***RTPSink中完成的)，所以主要分析的是从source中获取数据的过程。一个subsession对应一个流，当一个session中拥有多个subsession时，需要对每一个subsession进行单独的控制。我们可以看到，对处理“PLAY”命令时，对session中的每个subsession都调用了一次startStream操作，如下：

[cpp] view plain copy

print?

1. void RTSPServer::RTSPClientSession  
2.   ::handleCmd_PLAY(ServerMediaSubsession* subsession, char const* cseq,  
3.   char const* fullRequestStr) {  
4.   
5.   //现在终于开始媒体数据传输了  
6.   
7.   // Now, start streaming:  
8.   for (i = 0; i < fNumStreamStates; ++i) {  
9.     if (subsession == NULL /* means: aggregated operation */  
10. || subsession == fStreamStates[i].subsession) {  
11.       unsigned short rtpSeqNum = 0;  
12.       unsigned rtpTimestamp = 0;  
13.   
14.       //开始各个subsession上的数据传输, 即开始播放了  
15.   
16.       fStreamStates[i].subsession->startStream(fOurSessionId,  
17.       fStreamStates[i].streamToken,  
18.       (TaskFunc*)noteClientLiveness, this,  
19.       rtpSeqNum, rtpTimestamp,  
20.       handleAlternativeRequestByte, this);  
21.     ...  
22.     }  
23.   }  
24.     
25. ...  
26. }  

    
    PLAY命令的处理过程前面的文章已经分析过了，通过subsession->startStream调用启动每个流上的播放，其从source获取源数据是将在MultiFramedRTPSink::packFrame()中进行的。

[cpp] view plain copy

print?

1. void MultiFramedRTPSink::packFrame() {  
2.   if (fOutBuf->haveOverflowData()) {  
3. ...  
4.   } else {  
5. ...  
6.   
7.     //  
8.     //从source中获取下一个frame  
9.     //  
10.     fSource->getNextFrame(fOutBuf->curPtr(), fOutBuf->totalBytesAvailable(),  
11.               afterGettingFrame, this, ourHandleClosure, this);  
12.   }  
13. }  

    对于mpeg来讲，fSource是一个MPEG1or2VideoStreamFramer或者MPEG1or2AudioStreamFramer实例(根据live555中的源码，还有一种AAC格式音频，这里为了简便不作分析)。它们的继承关系如下：
    MPEG1or2VideoStreamFramer->MPEGVideoStreamFramer->FramedFilter->FramedSource->MediaSource
    MPEG1or2AudioStreamFramer->FramedSource->MediaSource


    先来分析mpeg video的处理。
    MPEG1or2VideoStreamFramer类的实现简单，主要是创建对应的语法分析器MPEG1or2VideoStreamParser实例。显然MPEG1or2VideoStreamFramer(是FrameSource的直接子类)是一个Filter，跟它的创建过程可以发现，它的输入source是一个MPEG1or2DemuxedElementaryStream类实例。对于单流的文件来讲, 一般包装的是ByteStreamFileSource类实例，从后面我们可以发现，最终直接读取文件的还是ByteStreamFileSource实例。对于语法分析部分，不作分析，我们只关心如何从文件中解析出音视频数据，所以直接跟踪Filter所包装的MPEG1or2DemuxedElementaryStream类。在语法分析器中，将会调用MPEG1or2DemuxedElementaryStream的getNextFrame函数。
    

    getNextFrame是定义在FramedSource中的非虚函数，实现如下

[cpp] view plain copy

print?

1. void FramedSource::getNextFrame(unsigned char* to, unsigned maxSize,  
2.                 afterGettingFunc* afterGettingFunc,  
3.                 void* afterGettingClientData,  
4.                 onCloseFunc* onCloseFunc,  
5.                 void* onCloseClientData) {  
6.   // Make sure we're not already being read:  
7.   if (fIsCurrentlyAwaitingData) {  
8.     envir() << "FramedSource[" << this << "]::getNextFrame(): attempting to read more than once at the same time!\n";  
9.     envir().internalError();  
10.   }  
11.   
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.   
23.   
24.   doGetNextFrame();     //获取下一个frame  
25. }  

    FramedSource::getNextFrame完成了一些员变量的初始化工作，其它工作交给dogetNextFrame函数，它是FrameSource上的一个纯虚函数，在子类MPEG1or2DemuxedElementaryStream中重新实现。
    

[cpp] view plain copy

print?

1. void MPEG1or2DemuxedElementaryStream::doGetNextFrame() {  
2.   fOurSourceDemux.getNextFrame(fOurStreamIdTag, fTo, fMaxSize,  
3.                    afterGettingFrame, this,  
4.                    handleClosure, this);  
5. }  

    fOurSourceDemux，被定义为MPEG1or2Demux(直接继承自Medium)对像的引用。为什么是引用而不是指针？我们想想，前面涉及到的类MPEG1or2DemuxedServerMediaSubsession、MPEG1or2VideoRTPSink、MPEG1or2VideoStreamFramer、MPEG1or2VideoStreamParser，这些对于文件中的每一个流(当然这里只是列出了视频相关的类)都需要创建一个实例，但是所有的流对应的文件却只有一个，最后读取文件的过程必然只有一份实现。文件中的每个流会对应一个MPEG1or2DemuxedElementaryStream实例，但是却对应同一个MPEG1or2Demux实例。这里用引用而不是指针，就是为了说明fOurSourceDemux并属于某个MPEG1or2DemuxedElementaryStream实例。


    在创建MPEG1or2Demux实例之前，会创建一个ByteStreamFileSource，来看MPEG1or2Demux::getNextFrame函数的定义

[cpp] view plain copy

print?

1. void MPEG1or2Demux::getNextFrame(u_int8_t streamIdTag,  
2.                  unsigned char* to, unsigned maxSize,  
3.                  FramedSource::afterGettingFunc* afterGettingFunc,  
4.                  void* afterGettingClientData,  
5.                  FramedSource::onCloseFunc* onCloseFunc,  
6.                  void* onCloseClientData) {  
7.   // First, check whether we have saved data for this stream id:  
8.   //  
9.   // 检查缓存中是否已经存在streamIdTag流的数据  
10.   //  
11.   if (useSavedData(streamIdTag, to, maxSize,  
12.            afterGettingFunc, afterGettingClientData)) {  
13.     return;  
14.   }  
15.       
16.   //注意，这里设置了回调用函数  
17.   // Then save the parameters of the specified stream id:  
18.   registerReadInterest(streamIdTag, to, maxSize,  
19.                afterGettingFunc, afterGettingClientData,  
20.                onCloseFunc, onCloseClientData);  
21.   
22.   
23.   // Next, if we're the only currently pending read, continue looking for data:  
24.   if (fNumPendingReads == 1 || fHaveUndeliveredData) {  
25.     fHaveUndeliveredData = 0;  
26.     continueReadProcessing();       //继续读取数据  
27.   } // otherwise the continued read processing has already been taken care of  
28. }  

    
    上面的代码中首先调用useSavedData函数，检查缓存中是否有相应流的数据存在，若是不存在，就只有再从文件中读取了。现在可以做一个猜测，缓存是必需的，文件中的不同流数据交错排列，而读取时一般会按顺序来读。现在要读取一个视频frame，但是当前文件指针处读取到的刚好是音视频frame，我们就需要将这个音频frame保存到缓存中，这样一直到读取视频数据为止。实际情况是不是这样呢?
    
    先来看MPEG1or2Demux::useSavedData函数的实现

[cpp] view plain copy

print?

1. Boolean MPEG1or2Demux::useSavedData(u_int8_t streamIdTag,  
2.                     unsigned char* to, unsigned maxSize,  
3.                     FramedSource::afterGettingFunc* afterGettingFunc,  
4.                     void* afterGettingClientData) {  
5.   struct OutputDescriptor& out = fOutput[streamIdTag];      //fOutput是一个缓存数组  
6.   //正常情况，缓存中没有数据，直接返回了  
7.   if (out.savedDataHead == NULL) return False; // common case  
8.   
9.   
10.   unsigned totNumBytesCopied = 0;  
11.   
12.   
13.   //  
14.   //从OutputDescriptor类型的缓存中读取全部数据  
15.   //  
16.   while (maxSize > 0 && out.savedDataHead != NULL) {  
17.     OutputDescriptor::SavedData& savedData = *(out.savedDataHead);  
18.     unsigned char* from = &savedData.data[savedData.numBytesUsed];  
19.     unsigned numBytesToCopy = savedData.dataSize - savedData.numBytesUsed;  
20.     if (numBytesToCopy > maxSize) numBytesToCopy = maxSize;   
21.     memmove(to, from, numBytesToCopy);  
22.     to += numBytesToCopy;  
23.     maxSize -= numBytesToCopy;  
24.     out.savedDataTotalSize -= numBytesToCopy;  
25.     totNumBytesCopied += numBytesToCopy;  
26.     savedData.numBytesUsed += numBytesToCopy;  
27.     if (savedData.numBytesUsed == savedData.dataSize) {  
28.       out.savedDataHead = savedData.next;  
29.       if (out.savedDataHead == NULL) out.savedDataTail = NULL;  
30.       savedData.next = NULL;  
31.       delete &savedData;  
32.     }  
33.   }  
34.   
35.   
36.   out.isCurrentlyActive = True;  
37.   if (afterGettingFunc != NULL) {  
38.     struct timeval presentationTime;  
39.     presentationTime.tv_sec = 0; presentationTime.tv_usec = 0; // should fix #####  
40.     (*afterGettingFunc)(afterGettingClientData, totNumBytesCopied,    
41.             0 /* numTruncatedBytes */, presentationTime,      
42.             0 /* durationInMicroseconds ?????#####*/);  
43.   }  
44.   return True;  
45. }  

    来分析一下上面的代码。缓存被定义为一个OutputDescriptor类型数组，看其定义
      OutputDescriptor_t fOutput[256];
    竟然定义了一个256个实例，不过也只有这个样才能使用流索引streamIdTag直接访问数组元素，streamIdTag应该小于256。每一个流将对应一个OutputDescriptor实例。实际数据保存在 OutputDescriptor::SavedData类型的链表中。


    现在继续来看MPEG1or2Demux::getNextFrame函数，我们需要知道是怎么分离出不同媒体流的，来看continueReadProcessing函数

[cpp] view plain copy

print?

1. void MPEG1or2Demux::continueReadProcessing() {  
2.   while (fNumPendingReads > 0) {  
3.     unsigned char acquiredStreamIdTag = fParser->parse();   //文件语法分析  
4.       
5.     if (acquiredStreamIdTag != 0) { //若未读取到所需要的数据，这个值为0  
6.         //我们从输入源获取到一个frame  
7.       // We were able to acquire a frame from the input.  
8.       struct OutputDescriptor& newOut = fOutput[acquiredStreamIdTag];  
9.       newOut.isCurrentlyAwaitingData = False;   //指示我们可以读下一个frame了，parse中会根据这个值判断是否正在处理的流  
10.       // indicates that we can be read again  
11.         // (This needs to be set before the 'after getting' call below,  
12.         //  in case it tries to read another frame)  
13.   
14.   
15.       //  
16.       //调用“after getting”函数  
17.       //  
18.       // Call our own 'after getting' function.  Because we're not a 'leaf'  
19.       // source, we can call this directly, without risking infinite recursion.  
20.       if (newOut.fAfterGettingFunc != NULL) {  
21.     (*newOut.fAfterGettingFunc)(newOut.afterGettingClientData,  
22.                     newOut.frameSize, 0 /* numTruncatedBytes */,  
23.                     newOut.presentationTime,  
24.                     0 /* durationInMicroseconds ?????#####*/);  
25.       --fNumPendingReads;  
26.       }  
27.     } else {  
28.       // We were unable to parse a complete frame from the input, because:  
29.       // - we had to read more data from the source stream, or  
30.       // - we found a frame for a stream that was being read, but whose  
31.       //   reader is not ready to get the frame right now, or  
32.       // - the source stream has ended.  
33.       break;  
34.     }  
35.   }  
36. }  

    MPEG1or2Demux::continueReadProcessing的实现似乎很眼熟悉啊！是的，其与MPEGVideoStreamFramer::continueReadProcessing()有些相似。两个函数中均调用了语法分析器进行语法分析，不过应注意前者的语法分析器是MPEGProgramStreamParser类实例，完成了对复合文件的解析，主要目的是从中分离出音视频流数据(其实就是demux过程)，而后者是一个MPEG1or2VideoStreamParser，对指定的流数据进一步分析。MPEGProgramStreamParser的数据源，是一个ByteStreamFileSource实例，它是MPEG1or2Demux构造函数中传递下来的参数。若未读取到所需要的数据，fParser->parse()将返回0。


    来看MPEGProgramStreamParser::parse函数实现

[cpp] view plain copy

print?

1. unsigned char MPEGProgramStreamParser::parse() {  
2.   unsigned char acquiredStreamTagId = 0;  
3.   
4.   
5.   try {  
6.     do {  
7.       switch (fCurrentParseState) {  
8.       case PARSING_PACK_HEADER: {  
9.     parsePackHeader();      //分析包头  
10.     break;  
11.       }  
12.       case PARSING_SYSTEM_HEADER: {  
13.     parseSystemHeader();    //分析系统头  
14.     break;  
15.       }  
16.       case PARSING_PES_PACKET: {  
17.     acquiredStreamTagId = parsePESPacket(); //分析流数据  
18.     break;  
19.       }  
20.       }  
21.     } while(acquiredStreamTagId == 0);  
22.   
23.   
24.     return acquiredStreamTagId;  
25.   } catch (int /*e*/) {  
26. #ifdef DEBUG  
27.     fprintf(stderr, "MPEGProgramStreamParser::parse() EXCEPTION (This is normal behavior - *not* an error)\n");  
28.     fflush(stderr);  
29. #endif  
30.     return 0;  // the parsing got interrupted  
31.   }  
32. }  

对于mpeg文件格式不熟悉，就不看了，只看获取流数据包的函数MPEGProgramStreamParser::parsePESPacket

[cpp] view plain copy

print?

1. unsigned char MPEGProgramStreamParser::parsePESPacket() {  
2. ...  
3.     //  
4.     //检查source是否正在等待这个流类型，如果是就把数据交给它  
5.     //  
6.     // Check whether our using source is interested in this stream type.  
7.     // If so, deliver the frame to him:  
8.     MPEG1or2Demux::OutputDescriptor_t& out = fUsingDemux->fOutput[stream_id];  
9.     if (out.isCurrentlyAwaitingData) {  
10.       unsigned numBytesToCopy;  
11.       if (PES_packet_length > out.maxSize) {  
12.         numBytesToCopy = out.maxSize;  
13.       } else {  
14.     numBytesToCopy = PES_packet_length;  
15.       }  
16.   
17.   
18.       getBytes(out.to, numBytesToCopy); //拷贝数据  
19.       out.frameSize = numBytesToCopy;  
20.   
21.   
22.       // set out.presentationTime later #####  
23.       acquiredStreamIdTag = stream_id;  
24.       PES_packet_length -= numBytesToCopy;0  
25.     } else if (out.isCurrentlyActive) {  
26.     //  
27.     //这个流是需要用到的，但是不是现在，当需要的时候我们才能把数据交出去  
28.     //  
29.         // Someone has been reading this stream, but isn't right now.  
30.       // We can't deliver this frame until he asks for it, so punt for now.  
31.       // The next time he asks for a frame, he'll get it.  
32.   
33.   
34.       restoreSavedParserState(); // so we read from the beginning next time  
35.       fUsingDemux->fHaveUndeliveredData = True;  
36.       throw READER_NOT_READY;   //抛出异常了  
37.     } else if (out.isPotentiallyReadable &&  
38.            out.savedDataTotalSize + PES_packet_length < 1000000 /*limit*/) {  
39.         //  
40.         //这个流也是需要用到的，只是当前没有读取而已，因此将其保存到缓存中(OutputDescriptor)  
41.         //  
42.       // Someone is interested in this stream, but hasn't begun reading it yet.  
43.       // Save this data, so that the reader will get it when he later asks for it.  
44.       unsigned char* buf = new unsigned char[PES_packet_length];  
45.       getBytes(buf, PES_packet_length);  
46.       MPEG1or2Demux::OutputDescriptor::SavedData* savedData  
47.     = new MPEG1or2Demux::OutputDescriptor::SavedData(buf, PES_packet_length);   //新建一个SavedData实例  
48.       if (out.savedDataHead == NULL) {  
49.     out.savedDataHead = out.savedDataTail = savedData;  
50.       } else {  
51.     out.savedDataTail->next = savedData;  
52.     out.savedDataTail = savedData;  
53.       }  
54.       out.savedDataTotalSize += PES_packet_length;  
55.       PES_packet_length = 0;  
56.     }  
57.     skipBytes(PES_packet_length);  
58.   }  
59.   
60.   
61.   // Check for another PES Packet next:  
62.   setParseState(PARSING_PES_PACKET);  
63.   
64.   
65.  return acquiredStreamIdTag;  
66. }  

    MPEGProgramStreamParser::parsePESPacket函数会读取流中的数据，若读取到的流数据不是当前需要的就保存到缓存中，也就是其对应的OutputDescriptor实例。需要注意局部变量acquiredStreamIdTag，它的值被初始化为0, 只有当读取到当前所需要的流数据时才会被赋值。所以若未读取到所需要的数据，这个函数将返回0值。