mpeg2transportstreammultiplexor.cpp

H.264 RTSP 串流(live 555)視窗版本

/**********
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.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
**********/
// "liveMedia"
// Copyright (c) 1996-2010 Live Networks, Inc.  All rights reserved.
// A class for generating MPEG-2 Transport Stream from one or more input
// Elementary Stream data sources
// Implementation

#include "MPEG2TransportStreamMultiplexor.hh"

#define TRANSPORT_PACKET_SIZE 188

#define PAT_FREQUENCY 100 // # of packets between Program Association Tables
#define PMT_FREQUENCY 500 // # of packets between Program Map Tables

#define PID_TABLE_SIZE 256

MPEG2TransportStreamMultiplexor
::MPEG2TransportStreamMultiplexor(UsageEnvironment& env)
  : FramedSource(env),
    fHaveVideoStreams(True/*by default*/),
    fOutgoingPacketCounter(0), fProgramMapVersion(0),
    fPreviousInputProgramMapVersion(0xFF), fCurrentInputProgramMapVersion(0xFF),
    fPCR_PID(0), fCurrentPID(0),
    fInputBuffer(NULL), fInputBufferSize(0), fInputBufferBytesUsed(0),
    fIsFirstAdaptationField(True) {
  for (unsigned i = 0; i < PID_TABLE_SIZE; ++i) {
    fPIDState[i].counter = 0;
    fPIDState[i].streamType = 0;
  }
}

MPEG2TransportStreamMultiplexor::~MPEG2TransportStreamMultiplexor() {
}

void MPEG2TransportStreamMultiplexor::doGetNextFrame() {
  if (fInputBufferBytesUsed >= fInputBufferSize) {
    // No more bytes are available from the current buffer.
    // Arrange to read a new one.
    awaitNewBuffer(fInputBuffer);
    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 MPEG2TransportStreamMultiplexor
::handleNewBuffer(unsigned char* buffer, unsigned bufferSize,
		  int mpegVersion, MPEG1or2Demux::SCR scr) {
  if (bufferSize < 4) return;
  fInputBuffer = buffer;
  fInputBufferSize = bufferSize;
  fInputBufferBytesUsed = 0;

  u_int8_t stream_id = fInputBuffer[3];
  // Use "stream_id" directly as our PID.
  // Also, figure out the Program Map 'stream type' from this.
  if (stream_id == 0xBE) { // padding_stream; ignore
    fInputBufferSize = 0;
  } else if (stream_id == 0xBC) { // program_stream_map
    setProgramStreamMap(fInputBufferSize);
    fInputBufferSize = 0; // then, ignore the buffer
  } else {
    fCurrentPID = stream_id;

    // 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&0xF0) == 0xE0) { // video
	streamType = mpegVersion == 1 ? 1 : mpegVersion == 2 ? 2 : mpegVersion == 4 ? 0x10 : 0x1B;
      } else if ((stream_id&0xE0) == 0xC0) { // audio
	streamType = mpegVersion == 1 ? 3 : mpegVersion == 2 ? 4 : 0xF;
      } else if (stream_id == 0xBD) { // private_stream1 (usually AC-3)
	streamType = 0x06; // for DVB; for ATSC, use 0x81
      } else { // something else, e.g., AC-3 uses private_stream1 (0xBD)
	streamType = 0x81; // private
      }
    }

    if (fPCR_PID == 0) { // set it to this stream, if it's appropriate:
      if ((!fHaveVideoStreams && (streamType == 3 || streamType == 4 || streamType == 0xF))/* audio stream */ ||
	  (streamType == 1 || streamType == 2 || streamType == 0x10 || streamType == 0x1B)/* video stream */) {
	fPCR_PID = fCurrentPID; // use this stream's SCR for PCR
      }
    }
    if (fCurrentPID == fPCR_PID) {
      // Record the input's current SCR timestamp, for use as our PCR:
      fPCR = scr;
    }
  }

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

void MPEG2TransportStreamMultiplexor
::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
      && !(fPCR.highBit == 0 && fPCR.remainingBits == 0 && fPCR.extension == 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) {
	u_int8_t flags = willAddPCR ? 0x10 : 0x00;
	if (fIsFirstAdaptationField) {
	  flags |= 0x80; // discontinuity_indicator
	  fIsFirstAdaptationField = False;
	}
	*header++ = flags;
	if (willAddPCR) {
	  u_int32_t pcrHigh32Bits = (fPCR.highBit<<31) | (fPCR.remainingBits>>1);
	  u_int8_t pcrLowBit = fPCR.remainingBits&1;
	  u_int8_t extHighBit = (fPCR.extension&0x100)>>8;
	  *header++ = pcrHigh32Bits>>24;
	  *header++ = pcrHigh32Bits>>16;
	  *header++ = pcrHigh32Bits>>8;
	  *header++ = pcrHigh32Bits;
	  *header++ = (pcrLowBit<<7)|0x7E|extHighBit;
	  *header++ = (u_int8_t)fPCR.extension; // low 8 bits of extension