mpeg1or2demux.cpp

流媒体传输协议的实现代码,非常有用.可以支持rtsp mms等流媒体传输协议

/**********
This library is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by the
Free Software Foundation; either version 2.1 of the License, or (at your
option) any later version. (See <http://www.gnu.org/copyleft/lesser.html>.)

This library is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for
more details.

You should have received a copy of the GNU Lesser General Public License
along with this library; if not, write to the Free Software Foundation, Inc.,
59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
**********/
// "liveMedia"
// Copyright (c) 1996-2004 Live Networks, Inc.  All rights reserved.
// Demultiplexer for a MPEG 1 or 2 Program Stream
// Implementation

#include "MPEG1or2Demux.hh"
#include "MPEG1or2DemuxedElementaryStream.hh"
#include "StreamParser.hh"
#include <stdlib.h>

////////// MPEGProgramStreamParser definition //////////

// An enum representing the current state of the parser:
enum MPEGParseState {
  PARSING_PACK_HEADER,
  PARSING_SYSTEM_HEADER,
  PARSING_PES_PACKET
}; 

class MPEGProgramStreamParser: public StreamParser {
public:
  MPEGProgramStreamParser(MPEG1or2Demux* usingSource, FramedSource* inputSource);
  virtual ~MPEGProgramStreamParser();

public:
  unsigned char parse();
      // returns the stream id of a stream for which a frame was acquired,
      // or 0 if no such frame was acquired.

private:
  void setParseState(MPEGParseState parseState);

  void parsePackHeader();
  void parseSystemHeader();
  unsigned char parsePESPacket(); // returns as does parse()

  Boolean isSpecialStreamId(unsigned char stream_id) const;
  // for PES packet header parsing

private:
  MPEG1or2Demux* fUsingSource;
  MPEGParseState fCurrentParseState;
};


////////// MPEG1or2Demux::OutputDescriptor::SavedData definition/implementation //////////

class MPEG1or2Demux::OutputDescriptor::SavedData {
public:
  SavedData(unsigned char* buf, unsigned size)
    : next(NULL), data(buf), dataSize(size), numBytesUsed(0) {
  }
  virtual ~SavedData() {
    delete[] data;
    delete next;
  }

  SavedData* next;
  unsigned char* data;
  unsigned dataSize, numBytesUsed;
};


////////// MPEG1or2Demux implementation //////////

MPEG1or2Demux
::MPEG1or2Demux(UsageEnvironment& env,
		FramedSource* inputSource, Boolean reclaimWhenLastESDies)
  : Medium(env),
    fInputSource(inputSource), fMPEGversion(0),
    fNextAudioStreamNumber(0), fNextVideoStreamNumber(0),
    fReclaimWhenLastESDies(reclaimWhenLastESDies), fNumOutstandingESs(0),
    fNumPendingReads(0), fHaveUndeliveredData(False) {
  fParser = new MPEGProgramStreamParser(this, inputSource);
  for (unsigned i = 0; i < 256; ++i) {
    fOutput[i].savedDataHead = fOutput[i].savedDataTail = NULL;
    fOutput[i].isPotentiallyReadable = False;
    fOutput[i].isCurrentlyActive = False;
    fOutput[i].isCurrentlyAwaitingData = False;
  } 
}

MPEG1or2Demux::~MPEG1or2Demux() {
  delete fParser;
  for (unsigned i = 0; i < 256; ++i) delete fOutput[i].savedDataHead;
  Medium::close(fInputSource);
}

MPEG1or2Demux* MPEG1or2Demux
::createNew(UsageEnvironment& env,
	    FramedSource* inputSource, Boolean reclaimWhenLastESDies) {
  // Need to add source type checking here???  #####

  return new MPEG1or2Demux(env, inputSource, reclaimWhenLastESDies);
}

MPEG1or2Demux::SCR::SCR()
  : highBit(0), remainingBits(0), extension(0), isValid(False) {
}

void MPEG1or2Demux
::noteElementaryStreamDeletion(MPEG1or2DemuxedElementaryStream* /*es*/) {
  if (--fNumOutstandingESs == 0 && fReclaimWhenLastESDies) {
    delete this;
  }
}

void MPEG1or2Demux::flushInput() {
  fParser->flushInput();
}

MPEG1or2DemuxedElementaryStream*
MPEG1or2Demux::newElementaryStream(u_int8_t streamIdTag) {
  ++fNumOutstandingESs;
  fOutput[streamIdTag].isPotentiallyReadable = True;
  return new MPEG1or2DemuxedElementaryStream(envir(), streamIdTag, *this);
}

MPEG1or2DemuxedElementaryStream* MPEG1or2Demux::newAudioStream() {
  unsigned char newAudioStreamTag = 0xC0 | (fNextAudioStreamNumber++&~0xE0);
      // MPEG audio stream tags are 110x xxxx (binary)
  return newElementaryStream(newAudioStreamTag);
}

MPEG1or2DemuxedElementaryStream* MPEG1or2Demux::newVideoStream() {
  unsigned char newVideoStreamTag = 0xE0 | (fNextVideoStreamNumber++&~0xF0);
      // MPEG video stream tags are 1110 xxxx (binary)
  return newElementaryStream(newVideoStreamTag);
}

// Appropriate one of the reserved stream id tags to mean: return raw PES packets:
#define RAW_PES 0xFC

MPEG1or2DemuxedElementaryStream* MPEG1or2Demux::newRawPESStream() {
  return newElementaryStream(RAW_PES);
}

void MPEG1or2Demux::registerReadInterest(u_int8_t streamIdTag,
				     unsigned char* to, unsigned maxSize,
				     FramedSource::afterGettingFunc* afterGettingFunc,
				     void* afterGettingClientData,
				     FramedSource::onCloseFunc* onCloseFunc,
				     void* onCloseClientData) {
  struct OutputDescriptor& out = fOutput[streamIdTag];
    
  // Make sure this stream is not already being read:
  if (out.isCurrentlyAwaitingData) {
    envir() << "MPEG1or2Demux::registerReadInterest(): attempt to read stream id "
	    << (void*)streamIdTag << " more than once!\n";
    exit(1);
  }
    
  out.to = to; out.maxSize = maxSize;
  out.fAfterGettingFunc = afterGettingFunc;
  out.afterGettingClientData = afterGettingClientData;
  out.fOnCloseFunc = onCloseFunc;
  out.onCloseClientData = onCloseClientData;
  out.isCurrentlyActive = True;
  out.isCurrentlyAwaitingData = True;
  // out.frameSize and out.presentationTime will be set when a frame's read

  ++fNumPendingReads;
}

Boolean MPEG1or2Demux::useSavedData(u_int8_t streamIdTag,
				    unsigned char* to, unsigned maxSize,
				    FramedSource::afterGettingFunc* afterGettingFunc,
				    void* afterGettingClientData) {
  struct OutputDescriptor& out = fOutput[streamIdTag];
  if (out.savedDataHead == NULL) return False; // common case

  unsigned totNumBytesCopied = 0;
  while (maxSize > 0 && out.savedDataHead != NULL) {
    OutputDescriptor::SavedData& savedData = *(out.savedDataHead);
    unsigned char* from = &savedData.data[savedData.numBytesUsed];
    unsigned numBytesToCopy = savedData.dataSize - savedData.numBytesUsed;
    if (numBytesToCopy > maxSize) numBytesToCopy = maxSize;
    memmove(to, from, numBytesToCopy);
    to += numBytesToCopy;
    maxSize -= numBytesToCopy;
    out.savedDataTotalSize -= numBytesToCopy;
    totNumBytesCopied += numBytesToCopy;
    savedData.numBytesUsed += numBytesToCopy;
    if (savedData.numBytesUsed == savedData.dataSize) {
      out.savedDataHead = savedData.next;
      if (out.savedDataHead == NULL) out.savedDataTail = NULL;
      savedData.next = NULL;
      delete &savedData;
    }
  }

  out.isCurrentlyActive = True;
  if (afterGettingFunc != NULL) {
    struct timeval presentationTime; // value? #####
    (*afterGettingFunc)(afterGettingClientData, totNumBytesCopied,
			0 /* numTruncatedBytes */, presentationTime,
			0 /* durationInMicroseconds ?????#####*/);
  }
  return True;
}

void MPEG1or2Demux
::continueReadProcessing(void* clientData,
			 unsigned char* /*ptr*/, unsigned /*size*/,
			 struct timeval /*presentationTime*/) {
  MPEG1or2Demux* demux = (MPEG1or2Demux*)clientData;
  demux->continueReadProcessing();
}

void MPEG1or2Demux::continueReadProcessing() {
  while (fNumPendingReads > 0) {
    unsigned char acquiredStreamIdTag = fParser->parse();

    if (acquiredStreamIdTag != 0) {
      // We were able to acquire a frame from the input.
      struct OutputDescriptor& newOut = fOutput[acquiredStreamIdTag]; 
      newOut.isCurrentlyAwaitingData = False;
        // indicates that we can be read again
        // (This needs to be set before the 'after getting' call below,
        //  in case it tries to read another frame)

      // Call our own 'after getting' function.  Because we're not a 'leaf'
      // source, we can call this directly, without risking infinite recursion.
      if (newOut.fAfterGettingFunc != NULL) {
	(*newOut.fAfterGettingFunc)(newOut.afterGettingClientData,
				    newOut.frameSize, 0 /* numTruncatedBytes */,
				    newOut.presentationTime,
				    0 /* durationInMicroseconds ?????#####*/);
      --fNumPendingReads;
      }
    } else {
      // We were unable to parse a complete frame from the input, because:
      // - we had to read more data from the source stream, or
      // - we found a frame for a stream that was being read, but whose
      //   reader is not ready to get the frame right now, or
      // - the source stream has ended.
      break;
    }
  }
}

void MPEG1or2Demux::getNextFrame(u_int8_t streamIdTag,
				 unsigned char* to, unsigned maxSize,
				 FramedSource::afterGettingFunc* afterGettingFunc,
				 void* afterGettingClientData,
				 FramedSource::onCloseFunc* onCloseFunc,
				 void* onCloseClientData) {
  // First, check whether we have saved data for this stream id:
  if (useSavedData(streamIdTag, to, maxSize,
		   afterGettingFunc, afterGettingClientData)) {
    return;
  }

  // Then save the parameters of the specified stream id:
  registerReadInterest(streamIdTag, to, maxSize,
		       afterGettingFunc, afterGettingClientData,
		       onCloseFunc, onCloseClientData);

  // Next, if we're the only currently pending read, continue looking for data:
  if (fNumPendingReads == 1 || fHaveUndeliveredData) {
    fHaveUndeliveredData = 0;
    continueReadProcessing();
  } // otherwise the continued read processing has already been taken care of
}

void MPEG1or2Demux::stopGettingFrames(u_int8_t streamIdTag) {
    struct OutputDescriptor& out = fOutput[streamIdTag];
    out.isCurrentlyActive = out.isCurrentlyAwaitingData = False;
}

void MPEG1or2Demux::handleClosure(void* clientData) {
  MPEG1or2Demux* demux = (MPEG1or2Demux*)clientData;

  demux->fNumPendingReads = 0;

  // Tell all pending readers that our source has closed.
  // Note that we need to make a copy of our readers' close functions
  // (etc.) before we start calling any of them, in case one of them
  // ends up deleting this.
  struct {
    FramedSource::onCloseFunc* fOnCloseFunc;
    void* onCloseClientData;
  } savedPending[256];
  unsigned i, numPending = 0;
  for (i = 0; i < 256; ++i) {
    struct OutputDescriptor& out = demux->fOutput[i];
    if (out.isCurrentlyAwaitingData) {
      if (out.fOnCloseFunc != NULL) {
	savedPending[numPending].fOnCloseFunc = out.fOnCloseFunc;
	savedPending[numPending].onCloseClientData = out.onCloseClientData;
	++numPending;
      }
    }
    delete out.savedDataHead; out.savedDataHead = out.savedDataTail = NULL;
    out.savedDataTotalSize = 0;
    out.isPotentiallyReadable = out.isCurrentlyActive = out.isCurrentlyAwaitingData
      = False;
  } 
  for (i = 0; i < numPending; ++i) {
    (*savedPending[i].fOnCloseFunc)(savedPending[i].onCloseClientData);
  }
}


////////// MPEGProgramStreamParser implementation //////////

#include <string.h>

MPEGProgramStreamParser::MPEGProgramStreamParser(MPEG1or2Demux* usingSource,
						 FramedSource* inputSource)
  : StreamParser(inputSource, MPEG1or2Demux::handleClosure, usingSource,
		 &MPEG1or2Demux::continueReadProcessing, usingSource),
  fUsingSource(usingSource), fCurrentParseState(PARSING_PACK_HEADER) {
}

MPEGProgramStreamParser::~MPEGProgramStreamParser() {
}

void MPEGProgramStreamParser::setParseState(MPEGParseState parseState) {
  fCurrentParseState = parseState;
  saveParserState();
}

unsigned char MPEGProgramStreamParser::parse() {
  unsigned char acquiredStreamTagId = 0;

  try {
    do {
      switch (fCurrentParseState) {
      case PARSING_PACK_HEADER: {
	parsePackHeader();
	break;
      }
      case PARSING_SYSTEM_HEADER: {
	parseSystemHeader();
	break;
      }
      case PARSING_PES_PACKET: {
	acquiredStreamTagId = parsePESPacket();
	break;
      }
      }
    } while(acquiredStreamTagId == 0);

    return acquiredStreamTagId;
  } catch (int /*e*/) {
#ifdef DEBUG
    fprintf(stderr, "MPEGProgramStreamParser::parse() EXCEPTION (This is normal behavior - *not* an error)\n");
    fflush(stderr);
#endif
    return 0;  // the parsing got interrupted
  }
}

#define PACK_START_CODE          0x000001BA
#define SYSTEM_HEADER_START_CODE 0x000001BB
#define PACKET_START_CODE_PREFIX 0x00000100

static inline Boolean isPacketStartCode(unsigned code) {
  return (code&0xFFFFFF00) == PACKET_START_CODE_PREFIX
    && code > SYSTEM_HEADER_START_CODE;
}