Parse program stream features out of transport streams

Add a program stream parser that aggregates program stream packets and re-emits them when the entire payload is available. Fix the test PES value to accurately use stuffing bytes. Update mp4 muxer page to exercise the TS parsing infrastructure.

@@ -14,9 +14,12 @@
 (function(window, videojs, undefined) {
 (function(window, videojs, undefined) {
 'use strict';
 'use strict';
 
 
-var PacketStream, ParseStream, MP2T_PACKET_LENGTH;
+var PacketStream, ParseStream, ProgramStream, Transmuxer, MP2T_PACKET_LENGTH, H264_STREAM_TYPE, ADTS_STREAM_TYPE, mp4;
 
 
 MP2T_PACKET_LENGTH = 188; // bytes
 MP2T_PACKET_LENGTH = 188; // bytes
+H264_STREAM_TYPE = 0x1b;
+ADTS_STREAM_TYPE = 0x0f;
+mp4 = videojs.mp4;
 
 
 /**
 /**
  * Splits an incoming stream of binary data into MP2T packets.
  * Splits an incoming stream of binary data into MP2T packets.
@@ -78,7 +81,7 @@ PacketStream.prototype = new videojs.Hls.Stream();
  */
  */
 ParseStream = function() {
 ParseStream = function() {
   var parsePsi, parsePat, parsePmt, parsePes, self;
   var parsePsi, parsePat, parsePmt, parsePes, self;
-  PacketStream.prototype.init.call(this);
+  ParseStream.prototype.init.call(this);
   self = this;
   self = this;
 
 
   this.programMapTable = {};
   this.programMapTable = {};
@@ -160,11 +163,12 @@ ParseStream = function() {
   };
   };
 
 
   parsePes = function(payload, pes) {
   parsePes = function(payload, pes) {
-    var ptsDtsFlags,
-        pesLength;
+    var ptsDtsFlags;
 
 
-    // PES packet length
-    pesLength = payload[4] << 8 | payload[5];
+    if (!pes.payloadUnitStartIndicator) {
+      pes.data = payload;
+      return;
+    }
 
 
     // find out if this packets starts a new keyframe
     // find out if this packets starts a new keyframe
     pes.dataAlignmentIndicator = (payload[6] & 0x04) !== 0;
     pes.dataAlignmentIndicator = (payload[6] & 0x04) !== 0;
@@ -224,10 +228,11 @@ ParseStream = function() {
     result.pid <<= 8;
     result.pid <<= 8;
     result.pid |= packet[2];
     result.pid |= packet[2];
 
 
-    // if an adaption field is present, its length is specified by
-    // the fifth byte of the PES header. The adaptation field is
-    // used to specify some forms of timing and control data that we
-    // do not currently use.
+    // if an adaption field is present, its length is specified by the
+    // fifth byte of the TS packet header. The adaptation field is
+    // used to add stuffing to PES packets that don't fill a complete
+    // TS packet, and to specify some forms of timing and control data
+    // that we do not currently use.
     if (((packet[3] & 0x30) >>> 4) > 0x01) {
     if (((packet[3] & 0x30) >>> 4) > 0x01) {
       offset += packet[offset] + 1;
       offset += packet[offset] + 1;
     }
     }
@@ -254,12 +259,129 @@ ParseStream.STREAM_TYPES  = {
   adts: 0x0f
   adts: 0x0f
 };
 };
 
 
+ProgramStream = function() {
+  var
+    // PES packet fragments
+    video = {
+      data: [],
+      size: 0
+    },
+    audio = {
+      data: [],
+      size: 0
+    },
+    flushStream = function(stream, type) {
+      var
+        event = {
+          type: type,
+          data: new Uint8Array(stream.size)
+        },
+        i = 0,
+        fragment;
+
+      // do nothing if there is no buffered data
+      if (!stream.data.length) {
+        return;
+      }
+
+      // reassemble the packet
+      while (stream.data.length) {
+        fragment = stream.data.shift();
+
+        event.data.set(fragment.data, i);
+        i += fragment.data.byteLength;
+      }
+      stream.size = 0;
+
+      self.trigger('data', event);
+    },
+    self;
+
+  ProgramStream.prototype.init.call(this);
+  self = this;
+
+  this.push = function(data) {
+    ({
+      pat: function() {
+        self.trigger('data', data);
+      },
+      pes: function() {
+        var stream, streamType;
+
+        if (data.streamType === H264_STREAM_TYPE) {
+          stream = video;
+          streamType = 'video';
+        } else {
+          stream = audio;
+          streamType = 'audio';
+        }
+
+        // if a new packet is starting, we can flush the completed
+        // packet
+        if (data.payloadUnitStartIndicator) {
+          flushStream(stream, streamType);
+        }
+
+        // buffer this fragment until we are sure we've received the
+        // complete payload
+        stream.data.push(data);
+        stream.size += data.data.byteLength;
+      },
+      pmt: function() {
+        self.trigger('data', data);
+      }
+    })[data.type]();
+  };
+
+  /**
+   * Flush any remaining input. Video PES packets may be of variable
+   * length. Normally, the start of a new video packet can trigger the
+   * finalization of the previous packet. That is not possible if no
+   * more video is forthcoming, however. In that case, some other
+   * mechanism (like the end of the file) has to be employed. When it is
+   * clear that no additional data is forthcoming, calling this method
+   * will flush the buffered packets.
+   */
+  this.end = function() {
+    flushStream(video, 'video');
+    flushStream(audio, 'audio');
+  };
+};
+ProgramStream.prototype = new videojs.Hls.Stream();
+
+Transmuxer = function() {
+  var self = this, packetStream, parseStream, programStream;
+  Transmuxer.prototype.init.call(this);
+
+  // set up the parsing pipeline
+  packetStream = new PacketStream();
+  parseStream = new ParseStream();
+  programStream = new ProgramStream();
+
+  packetStream.pipe(parseStream);
+  parseStream.pipe(programStream);
+
+  programStream.on('data', function(data) {
+    self.trigger('data', data);
+  });
+
+  // feed incoming data to the front of the parsing pipeline
+  this.push = function(data) {
+    packetStream.push(data);
+  };
+  // flush any buffered data
+  this.end = programStream.end;
+};
+Transmuxer.prototype = new videojs.Hls.Stream();
+
 window.videojs.mp2t = {
 window.videojs.mp2t = {
   PAT_PID: 0x0000,
   PAT_PID: 0x0000,
   MP2T_PACKET_LENGTH: MP2T_PACKET_LENGTH,
   MP2T_PACKET_LENGTH: MP2T_PACKET_LENGTH,
-  H264_STREAM_TYPE: 0x1b,
-  ADTS_STREAM_TYPE: 0x0f,
+  H264_STREAM_TYPE: H264_STREAM_TYPE,
+  ADTS_STREAM_TYPE: ADTS_STREAM_TYPE,
   PacketStream: PacketStream,
   PacketStream: PacketStream,
-  ParseStream: ParseStream
+  ParseStream: ParseStream,
+  ProgramStream: ProgramStream,
+  Transmuxer: Transmuxer
 };
 };
 })(window, window.videojs);
 })(window, window.videojs);

@@ -24,7 +24,18 @@ var
   PacketStream = videojs.mp2t.PacketStream,
   PacketStream = videojs.mp2t.PacketStream,
   packetStream,
   packetStream,
   ParseStream = videojs.mp2t.ParseStream,
   ParseStream = videojs.mp2t.ParseStream,
-  parseStream;
+  parseStream,
+  ProgramStream = videojs.mp2t.ProgramStream,
+  programStream,
+
+  MP2T_PACKET_LENGTH = videojs.mp2t.MP2T_PACKET_LENGTH,
+  H264_STREAM_TYPE = videojs.mp2t.H264_STREAM_TYPE,
+  ADTS_STREAM_TYPE = videojs.mp2t.ADTS_STREAM_TYPE,
+  packetize,
+
+  PAT,
+  PMT,
+  standalonePes;
 
 
 module('MP2T Packet Stream', {
 module('MP2T Packet Stream', {
   setup: function() {
   setup: function() {
@@ -91,9 +102,12 @@ test('buffers extra after multiple packets', function() {
   equal(188, datas[2].length, 'parsed the finel packet');
   equal(188, datas[2].length, 'parsed the finel packet');
 });
 });
 
 
-module('MP2T Parse Stream', {
+module('MP2T ParseStream', {
   setup: function() {
   setup: function() {
+    packetStream = new PacketStream();
     parseStream = new ParseStream();
     parseStream = new ParseStream();
+
+    packetStream.pipe(parseStream);
   }
   }
 });
 });
 
 
@@ -122,6 +136,21 @@ test('parses generic packet properties', function() {
   ok(packet.pid, 'parsed PID');
   ok(packet.pid, 'parsed PID');
 });
 });
 
 
+test('parses piped data events', function() {
+  var packet;
+  parseStream.on('data', function(data) {
+    packet = data;
+  });
+
+  parseStream.push(new Uint8Array([
+    0x47, // sync byte
+    // tei:0 pusi:1 tp:0 pid:0 0000 0000 0001 tsc:01 afc:10 cc:11 padding: 00
+    0x40, 0x01, 0x6c
+  ]));
+
+  ok(packet, 'parsed a packet');
+});
+
 test('parses a data packet with adaptation fields', function() {
 test('parses a data packet with adaptation fields', function() {
   var packet;
   var packet;
   parseStream.on('data', function(data) {
   parseStream.on('data', function(data) {
@@ -193,13 +222,7 @@ otherwise:
  | pn | pn | r pmp:5 | pmp |
  | pn | pn | r pmp:5 | pmp |
 */
 */
 
 
-test('parses the program map table pid from the program association table (PAT)', function() {
-  var packet;
-  parseStream.on('data', function(data) {
-    packet = data;
-  });
-
-  parseStream.push(new Uint8Array([
+PAT = [
   0x47, // sync byte
   0x47, // sync byte
   // tei:0 pusi:1 tp:0 pid:0 0000 0000 0000
   // tei:0 pusi:1 tp:0 pid:0 0000 0000 0000
   0x40, 0x00,
   0x40, 0x00,
@@ -217,19 +240,20 @@ test('parses the program map table pid from the program association table (PAT)'
   0x00, 0x10,
   0x00, 0x10,
   // crc32:0000 0000 0000 0000 0000 0000 0000 0000
   // crc32:0000 0000 0000 0000 0000 0000 0000 0000
   0x00, 0x00, 0x00, 0x00
   0x00, 0x00, 0x00, 0x00
-  ]));
-  ok(packet, 'parsed a packet');
-  strictEqual(0x0010, parseStream.pmtPid, 'parsed PMT pid');
-});
+];
 
 
-test('parse the elementary streams from a program map table', function() {
+test('parses the program map table pid from the program association table (PAT)', function() {
   var packet;
   var packet;
   parseStream.on('data', function(data) {
   parseStream.on('data', function(data) {
     packet = data;
     packet = data;
   });
   });
-  parseStream.pmtPid = 0x0010;
 
 
-  parseStream.push(new Uint8Array([
+  parseStream.push(new Uint8Array(PAT));
+  ok(packet, 'parsed a packet');
+  strictEqual(0x0010, parseStream.pmtPid, 'parsed PMT pid');
+});
+
+PMT = [
   0x47, // sync byte
   0x47, // sync byte
   // tei:0 pusi:1 tp:0 pid:0 0000 0010 0000
   // tei:0 pusi:1 tp:0 pid:0 0000 0010 0000
   0x40, 0x10,
   0x40, 0x10,
@@ -257,7 +281,16 @@ test('parse the elementary streams from a program map table', function() {
   0x00, 0x00,
   0x00, 0x00,
   // crc
   // crc
   0x00, 0x00, 0x00, 0x00
   0x00, 0x00, 0x00, 0x00
-  ]));
+];
+
+test('parse the elementary streams from a program map table', function() {
+  var packet;
+  parseStream.on('data', function(data) {
+    packet = data;
+  });
+  parseStream.pmtPid = 0x0010;
+
+  parseStream.push(new Uint8Array(PMT));
 
 
   ok(packet, 'parsed a packet');
   ok(packet, 'parsed a packet');
   ok(parseStream.programMapTable, 'parsed a program map');
   ok(parseStream.programMapTable, 'parsed a program map');
@@ -351,23 +384,44 @@ test('parses an elementary stream packet with a pts and dts', function() {
   equal(2 / 90, packet.dts, 'parsed the dts');
   equal(2 / 90, packet.dts, 'parsed the dts');
 });
 });
 
 
-test('parses an elementary stream packet without a pts or dts', function() {
-
-  var packet;
-  parseStream.on('data', function(data) {
-    packet = data;
-  });
-
-  parseStream.programMapTable = {
-    0x11: 0x1b // pid 0x11 is h264 data
-  };
-
-  parseStream.push(new Uint8Array([
+standalonePes = [
   0x47, // sync byte
   0x47, // sync byte
   // tei:0 pusi:1 tp:0 pid:0 0000 0001 0001
   // tei:0 pusi:1 tp:0 pid:0 0000 0001 0001
   0x40, 0x11,
   0x40, 0x11,
-    // tsc:01 afc:01 cc:0000
-    0x50,
+  // tsc:01 afc:11 cc:0000
+  0x70,
+  // afl:1010 1100
+  0xac,
+  // di:0 rai:0 espi:0 pf:0 of:0 spf:0 tpdf:0 afef:0
+  0x00,
+  // stuffing_bytes (171 bytes)
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+
+  0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+  0xff, 0xff, 0xff,
   // pscp:0000 0000 0000 0000 0000 0001
   // pscp:0000 0000 0000 0000 0000 0001
   0x00, 0x00, 0x01,
   0x00, 0x00, 0x01,
   // sid:0000 0000 ppl:0000 0000 0000 0101
   // sid:0000 0000 ppl:0000 0000 0000 0101
@@ -380,7 +434,21 @@ test('parses an elementary stream packet without a pts or dts', function() {
   0x00,
   0x00,
   // "data":1010 1111 0000 0001
   // "data":1010 1111 0000 0001
   0xaf, 0x01
   0xaf, 0x01
-  ]));
+];
+
+test('parses an elementary stream packet without a pts or dts', function() {
+
+  var packet;
+  parseStream.on('data', function(data) {
+    packet = data;
+  });
+
+  // pid 0x11 is h264 data
+  parseStream.programMapTable = {
+    0x11: H264_STREAM_TYPE
+  };
+
+  parseStream.push(new Uint8Array(standalonePes));
 
 
   ok(packet, 'parsed a packet');
   ok(packet, 'parsed a packet');
   equal('pes', packet.type, 'recognized a PES packet');
   equal('pes', packet.type, 'recognized a PES packet');
@@ -392,4 +460,153 @@ test('parses an elementary stream packet without a pts or dts', function() {
   ok(!packet.dts, 'did not parse a dts');
   ok(!packet.dts, 'did not parse a dts');
 });
 });
 
 
+module('MP2T ProgramStream', {
+  setup: function() {
+    programStream = new ProgramStream();
+  }
+});
+
+packetize = function(data) {
+  var packet = new Uint8Array(MP2T_PACKET_LENGTH);
+  packet.set(data);
+  return packet;
+};
+
+test('parses metadata events from PSI packets', function() {
+  var metadatas = 0, datas = 0;
+  programStream.on('data', function(data) {
+    if (data.type === 'pat' || data.type === 'pmt') {
+      metadatas++;
+    }
+    datas++;
+  });
+  programStream.push({
+    type: 'pat'
+  });
+  programStream.push({
+    type: 'pmt'
+  });
+
+  equal(2, datas, 'data fired');
+  equal(2, metadatas, 'metadata generated');
+});
+
+test('parses standalone program stream packets', function() {
+  var packets = [];
+  programStream.on('data', function(packet) {
+    packets.push(packet);
+  });
+  programStream.push({
+    type: 'pes',
+    data: new Uint8Array(19)
+  });
+  programStream.end();
+
+  equal(1, packets.length, 'built one packet');
+  equal('audio', packets[0].type, 'identified audio data');
+  equal(19, packets[0].data.byteLength, 'parsed the correct payload size');
+});
+
+test('aggregates program stream packets from the transport stream', function() {
+  var events = [];
+  programStream.on('data', function(event) {
+    events.push(event);
+  });
+
+  programStream.push({
+    type: 'pes',
+    streamType: H264_STREAM_TYPE,
+    data: new Uint8Array(7)
+  });
+  equal(0, events.length, 'buffers partial packets');
+
+  programStream.push({
+    type: 'pes',
+    streamType: H264_STREAM_TYPE,
+    data: new Uint8Array(13)
+  });
+  programStream.end();
+  equal(1, events.length, 'built one packet');
+  equal('video', events[0].type, 'identified video data');
+  equal(20, events[0].data.byteLength, 'concatenated transport packets');
+});
+
+test('buffers audio and video program streams individually', function() {
+  var events = [];
+  programStream.on('data', function(event) {
+    events.push(event);
+  });
+
+  programStream.push({
+    type: 'pes',
+    payloadUnitStartIndicator: true,
+    streamType: H264_STREAM_TYPE,
+    data: new Uint8Array(1)
+  });
+  programStream.push({
+    type: 'pes',
+    payloadUnitStartIndicator: true,
+    streamType: ADTS_STREAM_TYPE,
+    data: new Uint8Array(1)
+  });
+  equal(0, events.length, 'buffers partial packets');
+
+  programStream.push({
+    type: 'pes',
+    streamType: H264_STREAM_TYPE,
+    data: new Uint8Array(1)
+  });
+  programStream.push({
+    type: 'pes',
+    streamType: ADTS_STREAM_TYPE,
+    data: new Uint8Array(1)
+  });
+  programStream.end();
+  equal(2, events.length, 'parsed a complete packet');
+  equal('video', events[0].type, 'identified video data');
+  equal('audio', events[1].type, 'identified audio data');
+});
+
+test('flushes the buffered packets when a new one of that type is started', function() {
+  var packets = [];
+  programStream.on('data', function(packet) {
+    packets.push(packet);
+  });
+  programStream.push({
+    type: 'pes',
+    payloadUnitStartIndicator: true,
+    streamType: H264_STREAM_TYPE,
+    data: new Uint8Array(1)
+  });
+  programStream.push({
+    type: 'pes',
+    payloadUnitStartIndicator: true,
+    streamType: ADTS_STREAM_TYPE,
+    data: new Uint8Array(7)
+  });
+  programStream.push({
+    type: 'pes',
+    streamType: H264_STREAM_TYPE,
+    data: new Uint8Array(1)
+  });
+  equal(0, packets.length, 'buffers packets by type');
+
+  programStream.push({
+    type: 'pes',
+    payloadUnitStartIndicator: true,
+    streamType: H264_STREAM_TYPE,
+    data: new Uint8Array(1)
+  });
+  equal(1, packets.length, 'built one packet');
+  equal('video', packets[0].type, 'identified video data');
+  equal(2, packets[0].data.byteLength, 'concatenated packets');
+
+  programStream.end();
+  equal(3, packets.length, 'built tow more packets');
+  equal('video', packets[1].type, 'identified video data');
+  equal(1, packets[1].data.byteLength, 'parsed the video payload');
+  equal('audio', packets[2].type, 'identified audio data');
+  equal(7, packets[2].data.byteLength, 'parsed the audio payload');
+});
+
 })(window, window.videojs);
 })(window, window.videojs);