流是對數據生產，消費的一種抽象，今天先分析一下流基類的實現

const EE = require('events');
const util = require('util');
// 流的基類
function Stream() {
  EE.call(this);
}
// 繼承事件訂閱分發的能力
util.inherits(Stream, EE);

流的基類只提供了一個函數就是pipe。用于實現管道化。這個方法代碼比較多，分開說。

1 處理數據事件

 function ondata(chunk) {
    // 源流有數據到達，并且目的流可寫
    if (dest.writable) {
      // 目的流過載并且源流實現了pause方法，那就暫停可讀流的讀取操作，等待目的流觸發drain事件
      if (false === dest.write(chunk) && source.pause) {
        source.pause();
      }
    }
 }
  // 監聽data事件，可讀流有數據的時候，會觸發data事件
  source.on('data', ondata);

  function ondrain() {
    // 目的流可寫了，并且可讀流可讀，切換成自動讀取模式
    if (source.readable && source.resume) {
      source.resume();
    }
  }
  // 監聽drain事件，目的流可以消費數據了就會觸發該事件
  dest.on('drain', ondrain);

這是管道化時流控實現的地方，主要是利用了write返回值和drain事件。

流關閉/結束處理

 // 目的流不是標準輸出或標準錯誤，并且end不等于false
  if (!dest._isStdio && (!options || options.end !== false)) {
    // 源流沒有數據可讀了，執行end回調，告訴目的流，沒有數據可讀了
    source.on('end', onend);
    // 源流關閉了，執行close回調
    source.on('close', onclose);
  }
  // 兩個函數只會執行一次，也只會執行一個
  var didOnEnd = false;
  function onend() {
    if (didOnEnd) return;
    didOnEnd = true;
    // 執行目的流的end函數，說明寫數據完畢
    dest.end();
  }

  function onclose() {
    if (didOnEnd) return;
    didOnEnd = true;
    // 銷毀目的流
    if (typeof dest.destroy === 'function') dest.destroy();
  }

這里是處理源流結束和關閉后，通知目的流的邏輯。

錯誤處理和事件清除

  // remove all the event listeners that were added.
  function cleanup() {
    source.removeListener('data', ondata);
    dest.removeListener('drain', ondrain);

    source.removeListener('end', onend);
    source.removeListener('close', onclose);

    source.removeListener('error', onerror);
    dest.removeListener('error', onerror);

    source.removeListener('end', cleanup);
    source.removeListener('close', cleanup);

    dest.removeListener('close', cleanup);
  }

  function onerror(er) {
    // 出錯了，清除注冊的事件，包括正在執行的onerror函數
    cleanup();
    // 如果用戶沒有監聽流的error事件，則拋出錯誤，所以我們業務代碼需要監聽error事件
    if (EE.listenerCount(this, 'error') === 0) {
      throw er; // Unhandled stream error in pipe.
    }
  }
  // 監聽流的error事件
  source.on('error', onerror);
  dest.on('error', onerror);
  // 源流關閉或者沒有數據可讀時，清除注冊的事件
  source.on('end', cleanup);
  source.on('close', cleanup);
  // 目的流關閉了也清除他注冊的事件
  dest.on('close', cleanup);

這里主要是處理了error事件和流關閉/結束/出錯時清除訂閱的事件。這就是流基類的所有邏輯。