m3u8stream - npm Package Compare versions

Comparing version 0.3.0 to 0.4.0

lib/index.js

@@ -1,7 +0,8 @@
-const PassThrough = require('stream').PassThrough;
-const urlResolve  = require('url').resolve;
-const miniget     = require('miniget');
-const m3u8        = require('./m3u8-parser');
-const Queue       = require('./queue');
-const parseTime   = require('./parse-time');
+const PassThrough   = require('stream').PassThrough;
+const urlResolve    = require('url').resolve;
+const miniget       = require('miniget');
+const m3u8Parser    = require('./m3u8-parser');
+const DashMPDParser = require('./dash-mpd-parser');
+const Queue         = require('./queue');
+const parseTime     = require('./parse-time');
 
@@ -20,2 +21,9 @@
   const requestOptions = options.requestOptions;
+  const Parser = {
+    'm3u8': m3u8Parser,
+    'dash-mpd': DashMPDParser,
+  }[options.parser || 'm3u8'];
+  if (!Parser) {
+    throw new TypeError(`parser '${options.parser}' not supported`);
+  }
   let relativeBegin = typeof options.begin === 'string';
@@ -25,2 +33,3 @@ let begin = relativeBegin ?
     Math.max(options.begin - liveBuffer, 0) || 0;
+  let liveBegin = Date.now() - liveBuffer;
 
@@ -41,3 +50,3 @@ let currSegment;
   function onError(err) {
-    if (destroyed) { return; }
+    if (ended) { return; }
     stream.emit('error', err);
@@ -50,6 +59,7 @@ // Stop on any error.
   let refreshThreshold;
+  let minRefreshTime;
+  let refreshTimeout;
   let fetchingPlaylist = false;
-  let destroyed = false;
   let ended = false;
-  let lastPlaylistItems = new Set();
+  let lastRefresh;
 
@@ -60,5 +70,6 @@ function onQueuedEnd(err) {
       onError(err);
-    } else if (!fetchingPlaylist && !destroyed && !ended &&
+    } else if (!fetchingPlaylist && !ended &&
       requestQueue.tasks.length + requestQueue.active === refreshThreshold) {
-      refreshPlaylist();
+      let ms = Math.max(0, minRefreshTime - (Date.now() - lastRefresh));
+      refreshTimeout = setTimeout(refreshPlaylist, ms);
     } else if (ended && !requestQueue.tasks.length && !requestQueue.active) {
@@ -70,32 +81,30 @@ stream.end();
   let currPlaylist;
+  let lastSeq;
   function refreshPlaylist() {
     fetchingPlaylist = true;
+    lastRefresh = Date.now();
     currPlaylist = miniget(playlistURL, requestOptions);
     currPlaylist.on('error', onError);
-    const parser = currPlaylist.pipe(new m3u8());
-
-    let currTime, nextItemDuration;
-    parser.on('tag', (tagName, value) => {
-      switch (tagName) {
-        case 'EXT-X-PROGRAM-DATE-TIME':
-          currTime = new Date(value).getTime();
-          if (relativeBegin && begin >= 0) {
-            begin += currTime;
-          }
-          break;
-        case 'EXTINF':
-          nextItemDuration = Math.round(parseFloat(value.split(',')[0], 10) * 1000);
-          break;
-        case 'EXT-X-ENDLIST':
-          ended = true;
-          break;
+    const parser = currPlaylist.pipe(new Parser(options.id));
+    let starttime = null;
+    parser.on('starttime', (a) => {
+      starttime = a;
+      if (relativeBegin && begin >= 0) {
+        begin += starttime;
       }
     });
+    parser.on('endlist', () => { ended = true; });
+    parser.on('endearly', () => { currPlaylist.unpipe(parser); });
 
-    let currPlaylistItems = new Set();
-    function addItem(time, item) {
-      if (lastPlaylistItems.has(item)) { return; }
-      begin = time;
-      currPlaylistItems.add(item);
-      requestQueue.push(item, onQueuedEnd);
+    let addedItems = [];
+    let liveAddedItems = [];
+    function addItem(item, isLive) {
+      if (item.seq <= lastSeq) { return; }
+      lastSeq = item.seq;
+      begin = item.time;
+      requestQueue.push(item.url, onQueuedEnd);
+      addedItems.push(item);
+      if (isLive) {
+        liveAddedItems.push(item);
+      }
     }
@@ -105,14 +114,15 @@
     parser.on('item', (item) => {
-      if (!currTime || begin <= currTime) {
-        addItem(currTime, item);
+      item.time = starttime;
+      if (!starttime || begin <= item.time) {
+        addItem(item, liveBegin <= item.time);
       } else {
-        tailedItems.push([nextItemDuration, currTime, item]);
-        tailedItemsDuration += nextItemDuration;
+        tailedItems.push(item);
+        tailedItemsDuration += item.duration;
         // Only keep the last `liveBuffer` of items.
         while (tailedItems.length > 1 &&
-          tailedItemsDuration - tailedItems[0][0] > liveBuffer) {
-          tailedItemsDuration -= tailedItems.shift()[0];
+          tailedItemsDuration - tailedItems[0].duration > liveBuffer) {
+          tailedItemsDuration -= tailedItems.shift().duration;
         }
       }
-      currTime += nextItemDuration;
+      starttime += item.duration;
     });
@@ -122,14 +132,17 @@
       currPlaylist = null;
-      // If stream is behind by a bit, make sure to get the latest available
-      // items with a small buffer.
-      if (!currPlaylistItems.size && tailedItems.length) {
-        tailedItems.forEach((item) => {
-          addItem(item[1], item[2]);
-        });
+      // If we are too ahead of the stream, make sure to get the
+      // latest available items with a small buffer.
+      if (!addedItems.length && tailedItems.length) {
+        tailedItems.forEach((item) => { addItem(item, true); });
       }
 
       // Refresh the playlist when remaining segments get low.
-      refreshThreshold = Math.max(1, Math.ceil(currPlaylistItems.size * 0.01));
+      refreshThreshold = Math.max(1, Math.ceil(addedItems.length * 0.01));
+
+      // Throttle refreshing the playlist by looking at the duration
+      // of live items added on this refresh.
+      minRefreshTime =
+        addedItems.reduce(((total, item) => item.duration + total), 0);
+
       fetchingPlaylist = false;
-      lastPlaylistItems = currPlaylistItems;
     });
@@ -140,5 +153,6 @@ }
   stream.end = () => {
-    destroyed = true;
+    ended = true;
     streamQueue.die();
     requestQueue.die();
+    clearTimeout(refreshTimeout);
     if (currPlaylist) {
@@ -145,0 +159,0 @@ currPlaylist.unpipe();

lib/m3u8-parser.js

@@ -10,6 +10,8 @@ const Writable = require('stream').Writable;
  */
-module.exports = class m3u8parser extends Writable {
+module.exports = class m3u8Parser extends Writable {
   constructor() {
-    super({ decodeStrings: false });
+    super();
     this._lastLine = '';
+    this._seq = 0;
+    this._nextItemDuration = null;
     this.on('finish', () => {
@@ -22,10 +24,30 @@ this._parseLine(this._lastLine);
   _parseLine(line) {
-    let tag = line.match(/^#(EXT[A-Z0-9-]+)(?::(.*))?/);
-    if (tag) {
+    let match = line.match(/^#(EXT[A-Z0-9-]+)(?::(.*))?/);
+    if (match) {
       // This is a tag.
-      this.emit('tag', tag[1], tag[2] || null);
+      const tag = match[1];
+      const value = match[2] || null;
+      switch (tag) {
+        case 'EXT-X-PROGRAM-DATE-TIME':
+          this.emit('starttime', new Date(value).getTime());
+          break;
+        case 'EXT-X-MEDIA-SEQUENCE':
+          this._seq = parseInt(value, 10);
+          break;
+        case 'EXTINF':
+          this._nextItemDuration =
+            Math.round(parseFloat(value.split(',')[0], 10) * 1000);
+          break;
+        case 'EXT-X-ENDLIST':
+          this.emit('endlist');
+          break;
+      }
 
     } else if (!/^#/.test(line) && line.trim()) {
       // This is a segment
-      this.emit('item', line.trim());
+      this.emit('item', {
+        url: line.trim(),
+        seq: this._seq++,
+        duration: this._nextItemDuration,
+      });
     }
@@ -32,0 +54,0 @@ }

package.json

 {
   "name": "m3u8stream",
-  "description": "Concatenates segments from a m3u8 playlist into a consumable stream.",
+  "description": "Reads segments from a m3u8 or dash playlist into a consumable stream.",
   "keywords": [
     "m3u8",
+    "hls",
+    "dash",
     "live",
+    "playlist",
+    "segments",
     "stream"
   ],
-  "version": "0.3.0",
+  "version": "0.4.0",
   "repository": {
@@ -23,3 +27,4 @@ "type": "git",
   "dependencies": {
-    "miniget": "^1.1.0"
+    "miniget": "^1.1.0",
+    "sax": "^1.2.4"
   },
@@ -26,0 +31,0 @@ "devDependencies": {

README.md

 # node-m3u8stream
 
-Concatenates segments from a [m3u8 playlist](https://tools.ietf.org/html/draft-pantos-http-live-streaming-20) into a consumable stream.
+Reads segments from a [m3u8 playlist][1] or [DASH MPD file][2] into a consumable stream.
 
+[1]: https://tools.ietf.org/html/draft-pantos-http-live-streaming-20
+[2]: http://standards.iso.org/ittf/PubliclyAvailableStandards/MPEG-DASH_schema_files/DASH-MPD.xsd
+
 [![Build Status](https://secure.travis-ci.org/fent/node-m3u8stream.svg)](http://travis-ci.org/fent/node-m3u8stream)
@@ -32,2 +35,4 @@ [![Dependency Status](https://david-dm.org/fent/node-m3u8stream.svg)](https://david-dm.org/fent/node-m3u8stream)
 * `requestOptions` - Any options you want to pass to [miniget](https://github.com/fent/node-miniget), such as `headers`.
+* `parser` - Either "m3u8" or "dash-mpd". Defaults to "m3u8".
+* `id` - If playlist contains multiple media options. Otherwise, the first representation will be picked.
 
@@ -34,0 +39,0 @@ Stream has an `.end()` method, that if called, stops requesting segments, and refreshing the playlist.

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

16513

increased by41.95%

Number of package files

8

increased by14.29%

Lines of code

405

increased by57.59%

Number of lines in readme file

58

increased by9.43%

Worsened metrics

Dependency count

2

increased by100%

Dependency changes

• + Addedsax@^1.2.4

• + Addedsax@1.4.1(transitive)