mpeg2transportstreamfrompessource.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.
// A filter for converting a stream of MPEG PES packets to a MPEG-2 Transport Stream
// Implementation

#include "MPEG2TransportStreamFromPESSource.hh"

#define MAX_PES_PACKET_SIZE 65536
#define TRANSPORT_PACKET_SIZE 188

#define PAT_FREQUENCY 1000 // # of packets between Program Association Tables
#define PMT_FREQUENCY 5000 // # of packets between Program Map Tables

#define PID_TABLE_SIZE 256

MPEG2TransportStreamFromPESSource* MPEG2TransportStreamFromPESSource
::createNew(UsageEnvironment& env, MPEG1or2DemuxedElementaryStream* inputSource) {
  return new MPEG2TransportStreamFromPESSource(env, inputSource);
}

MPEG2TransportStreamFromPESSource
::MPEG2TransportStreamFromPESSource(UsageEnvironment& env,
				    MPEG1or2DemuxedElementaryStream* inputSource)
  : FramedFilter(env, inputSource),
    fOutgoingPacketCounter(0), fProgramMapVersion(0),
    fPreviousInputProgramMapVersion(0xFF), fCurrentInputProgramMapVersion(0xFF),
    fPCR_PID(0), fCurrentPID(0),
    fPCRHighBit(0), fPCRRemainingBits(0), fPCRExtension(0),
    fInputBufferSize(0), fInputBufferBytesUsed(0) {
  for (unsigned i = 0; i < PID_TABLE_SIZE; ++i) {
    fPIDState[i].counter = 0;
    fPIDState[i].streamType = 0;
  }

  fInputBuffer = new unsigned char[MAX_PES_PACKET_SIZE];
}

MPEG2TransportStreamFromPESSource::~MPEG2TransportStreamFromPESSource() {
  delete[] fInputBuffer;
}

void MPEG2TransportStreamFromPESSource::doGetNextFrame() {
  if (fInputBufferBytesUsed >= fInputBufferSize) {
    // No more bytes are available from the current PES packet.
    // Arrange to read a new one.
    fInputSource->getNextFrame(fInputBuffer, MAX_PES_PACKET_SIZE,
			       afterGettingFrame, this,
			       FramedSource::handleClosure, this);
    return;
  }

  do {
    // Periodically return a Program Association Table packet instead:
    if (fOutgoingPacketCounter++ % PAT_FREQUENCY == 0) {
      deliverPATPacket();
      break;
    }

    // Periodically (or when we see a new PID) return a Program Map Table instead:
    Boolean programMapHasChanged = fPIDState[fCurrentPID].counter == 0
      || fCurrentInputProgramMapVersion != fPreviousInputProgramMapVersion;
    if (fOutgoingPacketCounter % PMT_FREQUENCY == 0 || programMapHasChanged) {
      if (programMapHasChanged) { // reset values for next time:
	fPIDState[fCurrentPID].counter = 1;
	fPreviousInputProgramMapVersion = fCurrentInputProgramMapVersion;
      }
      deliverPMTPacket(programMapHasChanged);
      break;
    }

    // Normal case: Deliver (or continue delivering) the recently-read data:
    deliverDataToClient(fCurrentPID, fInputBuffer, fInputBufferSize,
			fInputBufferBytesUsed);
  } while (0);

  // NEED TO SET fPresentationTime, durationInMicroseconds #####
  // Complete the delivery to the client:
  afterGetting(this);
}

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

void MPEG2TransportStreamFromPESSource
::afterGettingFrame1(unsigned frameSize,
		     unsigned /*numTruncatedBytes*/,
		     struct timeval /*presentationTime*/,
		     unsigned /*durationInMicroseconds*/) {
  if (frameSize >= 4) {
    u_int8_t stream_id = fInputBuffer[3];
    // Use this stream_id directly as our PID.
    // Also, figure out the Program Map 'stream type' from this.
    if (stream_id == 0xBE) { // padding_stream; ignore
      frameSize = 0;
    } else if (stream_id == 0xBC) { // program_stream_map
      setProgramStreamMap(frameSize);
      frameSize = 0;
    } else {
      fCurrentPID = stream_id;
      if (fPCR_PID == 0) fPCR_PID = fCurrentPID; // use this stream's SCR for PCR

      MPEG1or2DemuxedElementaryStream* source
	= (MPEG1or2DemuxedElementaryStream*)fInputSource;

      // Set the stream's type:
      u_int8_t& streamType = fPIDState[fCurrentPID].streamType; // alias

      if (streamType == 0) {
	// Instead, set the stream's type to default values, based on whether
	// the stream is audio or video, and whether it's MPEG-1 or MPEG-2:
	if ((stream_id&0xE0) == 0xC0) { // audio
	  streamType = source->mpegVersion() == 1 ? 3 : 4;
	} else if ((stream_id&0xF0) == 0xE0) { // video
	  streamType = source->mpegVersion() == 1 ? 1 : 2;
	} else { // something else, e.g., AC-3 uses private_stream1 (0xBD)
	  streamType = 0x81; // private 
	}
      }

      if (fCurrentPID == fPCR_PID) {
	// Record the input's current SCR timestamp, for use as our PCR:
	fPCRHighBit = source->lastSeenSCR().highBit;
	fPCRRemainingBits = source->lastSeenSCR().remainingBits;
	fPCRExtension = source->lastSeenSCR().extension;
      }
    }
  }
  fInputBufferSize = frameSize;
  fInputBufferBytesUsed = 0;

  // Now that we have new PES packet data, retry the last delivery to the client:
  doGetNextFrame();
}

void MPEG2TransportStreamFromPESSource
::deliverDataToClient(u_int8_t pid, unsigned char* buffer, unsigned bufferSize,
		      unsigned& startPositionInBuffer) {
  // Construct a new Transport packet, and deliver it to the client:
  if (fMaxSize < TRANSPORT_PACKET_SIZE) {
    fFrameSize = 0; // the client hasn't given us enough space; deliver nothing
    fNumTruncatedBytes = TRANSPORT_PACKET_SIZE;
  } else {
    fFrameSize = TRANSPORT_PACKET_SIZE;
    Boolean willAddPCR = pid == fPCR_PID && startPositionInBuffer == 0;
    unsigned const numBytesAvailable = bufferSize - startPositionInBuffer;
    unsigned numHeaderBytes = 4; // by default
    unsigned numPCRBytes = 0; // by default
    unsigned numPaddingBytes = 0; // by default
    unsigned numDataBytes;
    u_int8_t adaptation_field_control;
    if (willAddPCR) {
      adaptation_field_control = 0x30;
      numHeaderBytes += 2; // for the "adaptation_field_length" and flags
      numPCRBytes = 6;
      if (numBytesAvailable >= TRANSPORT_PACKET_SIZE - numHeaderBytes - numPCRBytes) {
	numDataBytes = TRANSPORT_PACKET_SIZE - numHeaderBytes - numPCRBytes;
      } else {
	numDataBytes = numBytesAvailable;
	numPaddingBytes
	  = TRANSPORT_PACKET_SIZE - numHeaderBytes - numPCRBytes - numDataBytes;
      }
    } else if (numBytesAvailable >= TRANSPORT_PACKET_SIZE - numHeaderBytes) {
      // This is the common case
      adaptation_field_control = 0x10;
      numDataBytes = TRANSPORT_PACKET_SIZE - numHeaderBytes;
    } else {
      adaptation_field_control = 0x30;
      ++numHeaderBytes; // for the "adaptation_field_length"
      // ASSERT: numBytesAvailable <= TRANSPORT_PACKET_SIZE - numHeaderBytes
      numDataBytes = numBytesAvailable;
      if (numDataBytes < TRANSPORT_PACKET_SIZE - numHeaderBytes) {
	++numHeaderBytes; // for the adaptation field flags
	numPaddingBytes = TRANSPORT_PACKET_SIZE - numHeaderBytes - numDataBytes;
      }
    }
    // ASSERT: numHeaderBytes+numPCRBytes+numPaddingBytes+numDataBytes
    //         == TRANSPORT_PACKET_SIZE
    
    // Fill in the header of the Transport Stream packet:
    unsigned char* header = fTo;
    *header++ = 0x47; // sync_byte
    *header++ = (startPositionInBuffer == 0) ? 0x40 : 0x00;
      // transport_error_indicator, payload_unit_start_indicator, transport_priority,
      // first 5 bits of PID
    *header++ = pid;
      // last 8 bits of PID
    unsigned& continuity_counter = fPIDState[pid].counter; // alias
    *header++ = adaptation_field_control|(continuity_counter&0x0F);
      // transport_scrambling_control, adaptation_field_control, continuity_counter
    ++continuity_counter;
    if (adaptation_field_control == 0x30) {
      // Add an adaptation field:
      u_int8_t adaptation_field_length
	= (numHeaderBytes == 5) ? 0 : 1 + numPCRBytes + numPaddingBytes;
      *header++ = adaptation_field_length;
      if (numHeaderBytes > 5) {
	*header++ = willAddPCR ? 0x10 : 0x00; // various flags
	if (willAddPCR) {
	  u_int32_t pcrHigh32Bits = (fPCRHighBit<<31) | (fPCRRemainingBits>>1);
	  u_int8_t pcrLowBit = fPCRRemainingBits&1;