www.aleph1.co.uk Git - yaffs-website/blob - node_modules/ordered-read-streams/index.js

   1 var Readable = require('stream').Readable;
   2 var util = require('util');
   3
   4
   5 function addStream(streams, stream)
   6 {
   7   if(!stream.readable) throw new Error('All input streams must be readable');
   8
   9   if(this._readableState.ended) throw new Error('Adding streams after ended');
  10
  11
  12   var self = this;
  13
  14   stream._buffer = [];
  15
  16   stream.on('data', function(chunk)
  17   {
  18     if(this === streams[0])
  19       self.push(chunk);
  20
  21     else
  22       this._buffer.push(chunk);
  23   });
  24
  25   stream.on('end', function()
  26   {
  27     for(var stream = streams[0];
  28         stream && stream._readableState.ended;
  29         stream = streams[0])
  30     {
  31       while(stream._buffer.length)
  32         self.push(stream._buffer.shift());
  33
  34       streams.shift();
  35     }
  36
  37     if(!streams.length) self.push(null);
  38   });
  39
  40   stream.on('error', this.emit.bind(this, 'error'));
  41
  42
  43   streams.push(stream);
  44 }
  45
  46
  47 function OrderedStreams(streams, options) {
  48   if (!(this instanceof(OrderedStreams))) {
  49     return new OrderedStreams(streams, options);
  50   }
  51
  52   streams = streams || [];
  53   options = options || {};
  54
  55   options.objectMode = true;
  56
  57   Readable.call(this, options);
  58
  59
  60   if(!Array.isArray(streams)) streams = [streams];
  61   if(!streams.length) return this.push(null);  // no streams, close
  62
  63
  64   var addStream_bind = addStream.bind(this, []);
  65
  66
  67   this.concat = function()
  68   {
  69     Array.prototype.forEach.call(arguments, function(item)
  70     {
  71       if(Array.isArray(item))
  72         item.forEach(addStream_bind);
  73
  74       else
  75         addStream_bind(item);
  76     });
  77   };
  78
  79
  80   this.concat(streams);
  81 }
  82 util.inherits(OrderedStreams, Readable);
  83
  84 OrderedStreams.prototype._read = function () {};
  85
  86
  87 module.exports = OrderedStreams;