lib/stream.js

   1 'use strict';
   2
   3 module.exports = Stream;
   4
   5 const EE = require('events');
   6 const util = require('util');
   7
   8 util.inherits(Stream, EE);
   9 Stream.Readable = require('_stream_readable');
  10 Stream.Writable = require('_stream_writable');
  11 Stream.Duplex = require('_stream_duplex');
  12 Stream.Transform = require('_stream_transform');
  13 Stream.PassThrough = require('_stream_passthrough');
  14
  15 // Backwards-compat with node 0.4.x
  16 Stream.Stream = Stream;
  17
  18
  19 // old-style streams.  Note that the pipe method (the only relevant
  20 // part of this class) is overridden in the Readable class.
  21
  22 function Stream() {
  23   EE.call(this);
  24 }
  25
  26 Stream.prototype.pipe = function(dest, options) {
  27   var source = this;
  28
  29   function ondata(chunk) {
  30     if (dest.writable) {
  31       if (false === dest.write(chunk) && source.pause) {
  32         source.pause();
  33       }
  34     }
  35   }
  36
  37   source.on('data', ondata);
  38
  39   function ondrain() {
  40     if (source.readable && source.resume) {
  41       source.resume();
  42     }
  43   }
  44
  45   dest.on('drain', ondrain);
  46
  47   // If the 'end' option is not supplied, dest.end() will be called when
  48   // source gets the 'end' or 'close' events.  Only dest.end() once.
  49   if (!dest._isStdio && (!options || options.end !== false)) {
  50     source.on('end', onend);
  51     source.on('close', onclose);
  52   }
  53
  54   var didOnEnd = false;
  55   function onend() {
  56     if (didOnEnd) return;
  57     didOnEnd = true;
  58
  59     dest.end();
  60   }
  61
  62
  63   function onclose() {
  64     if (didOnEnd) return;
  65     didOnEnd = true;
  66
  67     if (typeof dest.destroy === 'function') dest.destroy();
  68   }
  69
  70   // don't leave dangling pipes when there are errors.
  71   function onerror(er) {
  72     cleanup();
  73     if (EE.listenerCount(this, 'error') === 0) {
  74       throw er; // Unhandled stream error in pipe.
  75     }
  76   }
  77
  78   source.on('error', onerror);
  79   dest.on('error', onerror);
  80
  81   // remove all the event listeners that were added.
  82   function cleanup() {
  83     source.removeListener('data', ondata);
  84     dest.removeListener('drain', ondrain);
  85
  86     source.removeListener('end', onend);
  87     source.removeListener('close', onclose);
  88
  89     source.removeListener('error', onerror);
  90     dest.removeListener('error', onerror);
  91
  92     source.removeListener('end', cleanup);
  93     source.removeListener('close', cleanup);
  94
  95     dest.removeListener('close', cleanup);
  96   }
  97
  98   source.on('end', cleanup);
  99   source.on('close', cleanup);
 100
 101   dest.on('close', cleanup);
 102
 103   dest.emit('pipe', source);
 104
 105   // Allow for unix-like usage: A.pipe(B).pipe(C)
 106   return dest;
 107 };