aboutsummaryrefslogtreecommitdiff
path: root/node_modules/ordered-read-streams/index.js
diff options
context:
space:
mode:
authorFlorian Dold <florian.dold@gmail.com>2016-10-10 03:43:44 +0200
committerFlorian Dold <florian.dold@gmail.com>2016-10-10 03:43:44 +0200
commitabd94a7f5a50f43c797a11b53549ae48fff667c3 (patch)
treeab8ed457f65cdd72e13e0571d2975729428f1551 /node_modules/ordered-read-streams/index.js
parenta0247c6a3fd6a09a41a7e35a3441324c4dcb58be (diff)
downloadwallet-core-abd94a7f5a50f43c797a11b53549ae48fff667c3.tar.xz
add node_modules to address #4364
Diffstat (limited to 'node_modules/ordered-read-streams/index.js')
-rw-r--r--node_modules/ordered-read-streams/index.js87
1 files changed, 87 insertions, 0 deletions
diff --git a/node_modules/ordered-read-streams/index.js b/node_modules/ordered-read-streams/index.js
new file mode 100644
index 000000000..f58ccf1dd
--- /dev/null
+++ b/node_modules/ordered-read-streams/index.js
@@ -0,0 +1,87 @@
+var Readable = require('stream').Readable;
+var util = require('util');
+
+
+function addStream(streams, stream)
+{
+ if(!stream.readable) throw new Error('All input streams must be readable');
+
+ if(this._readableState.ended) throw new Error('Adding streams after ended');
+
+
+ var self = this;
+
+ stream._buffer = [];
+
+ stream.on('data', function(chunk)
+ {
+ if(this === streams[0])
+ self.push(chunk);
+
+ else
+ this._buffer.push(chunk);
+ });
+
+ stream.on('end', function()
+ {
+ for(var stream = streams[0];
+ stream && stream._readableState.ended;
+ stream = streams[0])
+ {
+ while(stream._buffer.length)
+ self.push(stream._buffer.shift());
+
+ streams.shift();
+ }
+
+ if(!streams.length) self.push(null);
+ });
+
+ stream.on('error', this.emit.bind(this, 'error'));
+
+
+ streams.push(stream);
+}
+
+
+function OrderedStreams(streams, options) {
+ if (!(this instanceof(OrderedStreams))) {
+ return new OrderedStreams(streams, options);
+ }
+
+ streams = streams || [];
+ options = options || {};
+
+ options.objectMode = true;
+
+ Readable.call(this, options);
+
+
+ if(!Array.isArray(streams)) streams = [streams];
+ if(!streams.length) return this.push(null); // no streams, close
+
+
+ var addStream_bind = addStream.bind(this, []);
+
+
+ this.concat = function()
+ {
+ Array.prototype.forEach.call(arguments, function(item)
+ {
+ if(Array.isArray(item))
+ item.forEach(addStream_bind);
+
+ else
+ addStream_bind(item);
+ });
+ };
+
+
+ this.concat(streams);
+}
+util.inherits(OrderedStreams, Readable);
+
+OrderedStreams.prototype._read = function () {};
+
+
+module.exports = OrderedStreams;