m3u8stream
Advanced tools
Comparing version 0.1.3 to 0.2.0
@@ -13,3 +13,3 @@ const PassThrough = require('stream').PassThrough; | ||
*/ | ||
module.exports = function(playlistURL, options) { | ||
module.exports = (playlistURL, options) => { | ||
var stream = new PassThrough(); | ||
@@ -22,3 +22,3 @@ options = options || {}; | ||
var latestSegment; | ||
var streamQueue = new Queue(function(segment, callback) { | ||
var streamQueue = new Queue((segment, callback) => { | ||
latestSegment = segment; | ||
@@ -29,3 +29,3 @@ segment.pipe(stream, { end: false }); | ||
var requestQueue = new Queue(function(segmentURL, callback) { | ||
var requestQueue = new Queue((segmentURL, callback) => { | ||
var segment = miniget(urlResolve(playlistURL, segmentURL), requestOptions); | ||
@@ -36,3 +36,3 @@ segment.on('error', callback); | ||
concurrency: chunkReadahead, | ||
unique: function(segmentURL) { return segmentURL; }, | ||
unique: (segmentURL) => segmentURL, | ||
}); | ||
@@ -69,3 +69,3 @@ | ||
var parser = req.pipe(new m3u8()); | ||
parser.on('tag', function(tagName) { | ||
parser.on('tag', (tagName) => { | ||
if (tagName === 'EXT-X-ENDLIST') { | ||
@@ -78,7 +78,7 @@ ended = true; | ||
var totalItems = 0; | ||
parser.on('item', function(item) { | ||
parser.on('item', (item) => { | ||
totalItems++; | ||
requestQueue.push(item, onQueuedEnd); | ||
}); | ||
parser.on('end', function() { | ||
parser.on('end', () => { | ||
refreshThreshold = Math.ceil(totalItems * 0.01); | ||
@@ -91,3 +91,3 @@ tid = setTimeout(refreshPlaylist, refreshInterval); | ||
stream.end = function() { | ||
stream.end = () => { | ||
destroyed = true; | ||
@@ -94,0 +94,0 @@ streamQueue.die(); |
@@ -0,3 +1,4 @@ | ||
'use strict'; | ||
const Writable = require('stream').Writable; | ||
const util = require('util'); | ||
@@ -11,41 +12,37 @@ | ||
*/ | ||
var m3u8parser = module.exports = function() { | ||
var lastLine = ''; | ||
var self = this; | ||
function parseLine(line) { | ||
var tag = line.match(/^#(EXT[A-Z0-9\-]+)(?::(.*))?/); | ||
module.exports = class m3u8parser extends Writable { | ||
constructor() { | ||
super({ decodeStrings: false }); | ||
this._lastLine = ''; | ||
this.on('finish', () => { | ||
this._parseLine(this._lastLine); | ||
this.emit('end'); | ||
}); | ||
} | ||
_parseLine(line) { | ||
var tag = line.match(/^#(EXT[A-Z0-9-]+)(?::(.*))?/); | ||
if (tag) { | ||
// This is a tag. | ||
self.emit('tag', tag[1], tag[2] || null); | ||
this.emit('tag', tag[1], tag[2] || null); | ||
} else if (!/^#/.test(line) && line.trim()) { | ||
// This is a segment | ||
self.emit('item', line.trim()); | ||
this.emit('item', line.trim()); | ||
} | ||
} | ||
Writable.call(this, { | ||
decodeStrings: false, | ||
}), | ||
this._write = function(chunk, encoding, callback) { | ||
_write(chunk, encoding, callback) { | ||
var lines = chunk.toString('utf8').split('\n'); | ||
if (lastLine) { lines[0] = lastLine + lines[0]; } | ||
lines.forEach(function(line, i) { | ||
if (this._lastLine) { lines[0] = this._lastLine + lines[0]; } | ||
lines.forEach((line, i) => { | ||
if (i < lines.length - 1) { | ||
parseLine(line); | ||
this._parseLine(line); | ||
} else { | ||
// Save the last line in case it has been broken up. | ||
lastLine = line; | ||
this._lastLine = line; | ||
} | ||
}); | ||
callback(); | ||
}; | ||
this.on('finish', function() { | ||
parseLine(lastLine); | ||
self.emit('end'); | ||
}); | ||
} | ||
}; | ||
util.inherits(m3u8parser, Writable); |
114
lib/queue.js
@@ -1,63 +0,67 @@ | ||
/** | ||
* A really simple queue with concurrency that optionally only adds unique tasks. | ||
* | ||
* @param {Function(Object, Function)} worker | ||
* @param {Object} options | ||
*/ | ||
var Queue = module.exports = function(worker, options) { | ||
this._worker = worker; | ||
options = options || {}; | ||
this._concurrency = options.concurrency || 1; | ||
this._unique = options.unique; | ||
this._tasksMap = {}; | ||
this.tasks = []; | ||
this.active = 0; | ||
}; | ||
'use strict'; | ||
module.exports = class Queue { | ||
/** | ||
* A really simple queue with concurrency that optionally | ||
* only adds unique tasks. | ||
* | ||
* @param {Function(Object, Function)} worker | ||
* @param {Object} options | ||
*/ | ||
constructor(worker, options) { | ||
this._worker = worker; | ||
options = options || {}; | ||
this._concurrency = options.concurrency || 1; | ||
this._unique = options.unique; | ||
this._tasksMap = {}; | ||
this.tasks = []; | ||
this.active = 0; | ||
} | ||
/** | ||
* Push a task to the queue. | ||
* | ||
* @param {Object} item | ||
* @param {Function(Error)} callback | ||
*/ | ||
Queue.prototype.push = function(item) { | ||
if (this._unique) { | ||
var key = this._unique(item); | ||
if (this._tasksMap[key] === true) { return; } | ||
this._tasksMap[key] = true; | ||
/** | ||
* Push a task to the queue. | ||
* | ||
* @param {Object} item | ||
* @param {Function(Error)} callback | ||
*/ | ||
push(item) { | ||
if (this._unique) { | ||
var key = this._unique(item); | ||
if (this._tasksMap[key] === true) { return; } | ||
this._tasksMap[key] = true; | ||
} | ||
this.tasks.push(arguments); | ||
this._next(); | ||
} | ||
this.tasks.push(arguments); | ||
this._next(); | ||
}; | ||
/** | ||
* Process next job in queue. | ||
*/ | ||
Queue.prototype._next = function() { | ||
if (this.active >= this._concurrency || !this.tasks.length) { return; } | ||
var task = this.tasks.shift(); | ||
var item = task[0]; | ||
var callback = task[1]; | ||
var callbackCalled = false; | ||
var self = this; | ||
this.active++; | ||
this._worker(item, function(err) { | ||
if (callbackCalled) { return; } | ||
if (self._unique) { delete self._tasksMap[self._unique(item)]; } | ||
self.active--; | ||
callbackCalled = true; | ||
if (callback) { callback(err); } | ||
self._next(); | ||
}); | ||
}; | ||
/** | ||
* Process next job in queue. | ||
*/ | ||
_next() { | ||
if (this.active >= this._concurrency || !this.tasks.length) { return; } | ||
var task = this.tasks.shift(); | ||
var item = task[0]; | ||
var callback = task[1]; | ||
var callbackCalled = false; | ||
this.active++; | ||
this._worker(item, (err) => { | ||
if (callbackCalled) { return; } | ||
if (this._unique) { delete this._tasksMap[this._unique(item)]; } | ||
this.active--; | ||
callbackCalled = true; | ||
if (callback) { callback(err); } | ||
this._next(); | ||
}); | ||
} | ||
/** | ||
* Stops processing queued jobs. | ||
*/ | ||
Queue.prototype.die = function() { | ||
this.tasks = []; | ||
this._tasksMap = {}; | ||
/** | ||
* Stops processing queued jobs. | ||
*/ | ||
die() { | ||
this.tasks = []; | ||
this._tasksMap = {}; | ||
} | ||
}; |
@@ -9,3 +9,3 @@ { | ||
], | ||
"version": "0.1.3", | ||
"version": "0.2.0", | ||
"repository": { | ||
@@ -28,10 +28,10 @@ "type": "git", | ||
"istanbul": "^0.4.5", | ||
"mocha": "^3.2.0", | ||
"mocha": "^4.0.0", | ||
"nock": "^9.0.9", | ||
"sinon": "^2.0.0" | ||
"sinon": "^4.0.0" | ||
}, | ||
"engines": { | ||
"node": ">=0.12" | ||
"node": ">=4" | ||
}, | ||
"license": "MIT" | ||
} |
License Policy Violation
LicenseThis package is not allowed per your license policy. Review the package's license to ensure compliance.
Found 1 instance in 1 package
License Policy Violation
LicenseThis package is not allowed per your license policy. Review the package's license to ensure compliance.
Found 1 instance in 1 package
187
9174