-
-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
BREAKING: Convert to ES class to drop
inherits dep
Replace `MultiStream()` with `new MultiStream()` now. Readme examples have always shown this usage, but now it is required. For: brave/brave-browser#5490
- Loading branch information
Showing
4 changed files
with
117 additions
and
125 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,152 +1,145 @@ | ||
| module.exports = MultiStream | ||
|
|
||
| var inherits = require('inherits') | ||
| var stream = require('readable-stream') | ||
|
|
||
| inherits(MultiStream, stream.Readable) | ||
|
|
||
| function MultiStream (streams, opts) { | ||
| var self = this | ||
| if (!(self instanceof MultiStream)) return new MultiStream(streams, opts) | ||
| stream.Readable.call(self, opts) | ||
| function toStreams2Obj (s) { | ||
| return toStreams2(s, { objectMode: true, highWaterMark: 16 }) | ||
| } | ||
|
|
||
| self.destroyed = false | ||
| function toStreams2Buf (s) { | ||
| return toStreams2(s) | ||
| } | ||
|
|
||
| self._drained = false | ||
| self._forwarding = false | ||
| self._current = null | ||
| self._toStreams2 = (opts && opts.objectMode) ? toStreams2Obj : toStreams2Buf | ||
| function toStreams2 (s, opts) { | ||
| if (!s || typeof s === 'function' || s._readableState) return s | ||
|
|
||
| if (typeof streams === 'function') { | ||
| self._queue = streams | ||
| } else { | ||
| self._queue = streams.map(self._toStreams2) | ||
| self._queue.forEach(function (stream) { | ||
| if (typeof stream !== 'function') self._attachErrorListener(stream) | ||
| }) | ||
| var wrap = new stream.Readable(opts).wrap(s) | ||
| if (s.destroy) { | ||
| wrap.destroy = s.destroy.bind(s) | ||
| } | ||
|
|
||
| self._next() | ||
| return wrap | ||
| } | ||
|
|
||
| MultiStream.obj = function (streams) { | ||
| return new MultiStream(streams, { objectMode: true, highWaterMark: 16 }) | ||
| } | ||
| class MultiStream extends stream.Readable { | ||
| constructor (streams, opts) { | ||
| super(opts) | ||
|
|
||
| MultiStream.prototype._read = function () { | ||
| this._drained = true | ||
| this._forward() | ||
| } | ||
| this.destroyed = false | ||
|
|
||
| MultiStream.prototype._forward = function () { | ||
| if (this._forwarding || !this._drained || !this._current) return | ||
| this._forwarding = true | ||
| this._drained = false | ||
| this._forwarding = false | ||
| this._current = null | ||
| this._toStreams2 = (opts && opts.objectMode) ? toStreams2Obj : toStreams2Buf | ||
|
|
||
| var chunk | ||
| while ((chunk = this._current.read()) !== null && this._drained) { | ||
| this._drained = this.push(chunk) | ||
| } | ||
|
|
||
| this._forwarding = false | ||
| } | ||
| if (typeof streams === 'function') { | ||
| this._queue = streams | ||
| } else { | ||
| this._queue = streams.map(this._toStreams2) | ||
| this._queue.forEach(stream => { | ||
| if (typeof stream !== 'function') this._attachErrorListener(stream) | ||
| }) | ||
| } | ||
|
|
||
| MultiStream.prototype.destroy = function (err) { | ||
| if (this.destroyed) return | ||
| this.destroyed = true | ||
| this._next() | ||
| } | ||
|
|
||
| if (this._current && this._current.destroy) this._current.destroy() | ||
| if (typeof this._queue !== 'function') { | ||
| this._queue.forEach(function (stream) { | ||
| if (stream.destroy) stream.destroy() | ||
| }) | ||
| _read () { | ||
| this._drained = true | ||
| this._forward() | ||
| } | ||
|
|
||
| if (err) this.emit('error', err) | ||
| this.emit('close') | ||
| } | ||
| _forward () { | ||
| if (this._forwarding || !this._drained || !this._current) return | ||
| this._forwarding = true | ||
|
|
||
| MultiStream.prototype._next = function () { | ||
| var self = this | ||
| self._current = null | ||
|
|
||
| if (typeof self._queue === 'function') { | ||
| self._queue(function (err, stream) { | ||
| if (err) return self.destroy(err) | ||
| stream = self._toStreams2(stream) | ||
| self._attachErrorListener(stream) | ||
| self._gotNextStream(stream) | ||
| }) | ||
| } else { | ||
| var stream = self._queue.shift() | ||
| if (typeof stream === 'function') { | ||
| stream = self._toStreams2(stream()) | ||
| self._attachErrorListener(stream) | ||
| var chunk | ||
| while ((chunk = this._current.read()) !== null && this._drained) { | ||
| this._drained = this.push(chunk) | ||
| } | ||
| self._gotNextStream(stream) | ||
| } | ||
| } | ||
|
|
||
| MultiStream.prototype._gotNextStream = function (stream) { | ||
| var self = this | ||
|
|
||
| if (!stream) { | ||
| self.push(null) | ||
| self.destroy() | ||
| return | ||
| this._forwarding = false | ||
| } | ||
|
|
||
| self._current = stream | ||
| self._forward() | ||
| destroy (err) { | ||
| if (this.destroyed) return | ||
| this.destroyed = true | ||
|
|
||
| stream.on('readable', onReadable) | ||
| stream.once('end', onEnd) | ||
| stream.once('close', onClose) | ||
| if (this._current && this._current.destroy) this._current.destroy() | ||
| if (typeof this._queue !== 'function') { | ||
| this._queue.forEach(stream => { | ||
| if (stream.destroy) stream.destroy() | ||
| }) | ||
| } | ||
|
|
||
| function onReadable () { | ||
| self._forward() | ||
| if (err) this.emit('error', err) | ||
| this.emit('close') | ||
| } | ||
|
|
||
| function onClose () { | ||
| if (!stream._readableState.ended) { | ||
| self.destroy() | ||
| _next () { | ||
| this._current = null | ||
|
|
||
| if (typeof this._queue === 'function') { | ||
| this._queue((err, stream) => { | ||
| if (err) return this.destroy(err) | ||
| stream = this._toStreams2(stream) | ||
| this._attachErrorListener(stream) | ||
| this._gotNextStream(stream) | ||
| }) | ||
| } else { | ||
| var stream = this._queue.shift() | ||
| if (typeof stream === 'function') { | ||
| stream = this._toStreams2(stream()) | ||
| this._attachErrorListener(stream) | ||
| } | ||
| this._gotNextStream(stream) | ||
| } | ||
| } | ||
|
|
||
| function onEnd () { | ||
| self._current = null | ||
| stream.removeListener('readable', onReadable) | ||
| stream.removeListener('end', onEnd) | ||
| stream.removeListener('close', onClose) | ||
| self._next() | ||
| } | ||
| } | ||
| _gotNextStream (stream) { | ||
| if (!stream) { | ||
| this.push(null) | ||
| this.destroy() | ||
| return | ||
| } | ||
|
|
||
| MultiStream.prototype._attachErrorListener = function (stream) { | ||
| var self = this | ||
| if (!stream) return | ||
| this._current = stream | ||
| this._forward() | ||
|
|
||
| stream.once('error', onError) | ||
| const onReadable = () => { | ||
| this._forward() | ||
| } | ||
|
|
||
| function onError (err) { | ||
| stream.removeListener('error', onError) | ||
| self.destroy(err) | ||
| } | ||
| } | ||
| const onClose = () => { | ||
| if (!stream._readableState.ended) { | ||
| this.destroy() | ||
| } | ||
| } | ||
|
|
||
| function toStreams2Obj (s) { | ||
| return toStreams2(s, { objectMode: true, highWaterMark: 16 }) | ||
| } | ||
| const onEnd = () => { | ||
| this._current = null | ||
| stream.removeListener('readable', onReadable) | ||
| stream.removeListener('end', onEnd) | ||
| stream.removeListener('close', onClose) | ||
| this._next() | ||
| } | ||
|
|
||
| function toStreams2Buf (s) { | ||
| return toStreams2(s) | ||
| } | ||
| stream.on('readable', onReadable) | ||
| stream.once('end', onEnd) | ||
| stream.once('close', onClose) | ||
| } | ||
|
|
||
| function toStreams2 (s, opts) { | ||
| if (!s || typeof s === 'function' || s._readableState) return s | ||
| _attachErrorListener (stream) { | ||
| if (!stream) return | ||
|
|
||
| var wrap = new stream.Readable(opts).wrap(s) | ||
| if (s.destroy) { | ||
| wrap.destroy = s.destroy.bind(s) | ||
| const onError = (err) => { | ||
| stream.removeListener('error', onError) | ||
| this.destroy(err) | ||
| } | ||
|
|
||
| stream.once('error', onError) | ||
| } | ||
| return wrap | ||
| } | ||
|
|
||
| MultiStream.obj = streams => ( | ||
| new MultiStream(streams, { objectMode: true, highWaterMark: 16 }) | ||
| ) | ||
|
|
||
| module.exports = MultiStream |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters