1 var Readable = require('stream').Readable;
2 var util = require('util');
5 function addStream(streams, stream)
7 if(!stream.readable) throw new Error('All input streams must be readable');
9 if(this._readableState.ended) throw new Error('Adding streams after ended');
16 stream.on('data', function(chunk)
18 if(this === streams[0])
22 this._buffer.push(chunk);
25 stream.on('end', function()
27 for(var stream = streams[0];
28 stream && stream._readableState.ended;
31 while(stream._buffer.length)
32 self.push(stream._buffer.shift());
37 if(!streams.length) self.push(null);
40 stream.on('error', this.emit.bind(this, 'error'));
47 function OrderedStreams(streams, options) {
48 if (!(this instanceof(OrderedStreams))) {
49 return new OrderedStreams(streams, options);
52 streams = streams || [];
53 options = options || {};
55 options.objectMode = true;
57 Readable.call(this, options);
60 if(!Array.isArray(streams)) streams = [streams];
61 if(!streams.length) return this.push(null); // no streams, close
64 var addStream_bind = addStream.bind(this, []);
67 this.concat = function()
69 Array.prototype.forEach.call(arguments, function(item)
71 if(Array.isArray(item))
72 item.forEach(addStream_bind);
82 util.inherits(OrderedStreams, Readable);
84 OrderedStreams.prototype._read = function () {};
87 module.exports = OrderedStreams;