through2源碼學習及ts版

transform stream

through2 是一個 transform stream 的封裝庫，用來處理 node 的 stream，源碼雖然僅僅只有100多行，可是裏面的內容確實頗有看頭的！

Transform 概念

Transform 是一個變換流，既可讀也可寫 是雙工流 Duplex 的特殊形式。Duplex 的寫和讀沒有任何關聯，兩個緩衝區和管道互不干擾，而 Transform 將其輸入和輸出是存在相互關聯，經過轉換函數將流進行轉換。

transform stream 簡單案例

實現一個用來將目標字符串替換成指定字符串的變換流：

// replaceStream.ts

import { Transform } from "stream";

export default class ReplaceStream extends Transform {
  constructor(
    private searchString: string,
    private replaceString: string,
    private tailPiece: string = ""
  ) {
    super();
  }
  _transform(chunk, encoding, callback) {
    const pieces = (this.tailPiece + chunk).split(this.searchString);
    const lastPiece = pieces[pieces.length - 1];
    const tailPieceLen = this.searchString.length - 1;

    this.tailPiece = lastPiece.slice(-tailPieceLen);
    pieces[pieces.length - 1] = lastPiece.slice(0, -tailPieceLen);
    this.push(pieces.join(this.replaceString));

    callback();
  }

  /** 某些狀況下，轉換操做可能須要在流的末尾發送一些額外的數據。 */
  _flush(callback) {
    this.push(this.tailPiece);
    this.push("\n")
    this.push("haha")
    callback();
  }
}


// replaceStreamTest.ts

import ReplaceStream from "./replaceStream";

const re = new ReplaceStream("World", "Nodejs");

re.on("data", chunk => console.log(chunk.toString()));

re.write("hello w");
re.write("orld");
re.end();

複製代碼

建立一個新的類並繼承 stream.Transform 這個基類。該類的構造函數接受兩個參數：searchString 和 replaceString。這兩個參數用於指定須要查找匹配的字符串以及用來替換的字符串。同時還初始化了一個內部變量 tailPieceLen 提供給 _transform() 方法使用。

_transform() 方法和 _write() 方法有相同的方法簽名，但並非將數據直接寫到底層資源，而是使用 this.push() 方法將其推送到內部緩存，就像咱們在可讀流 _read() 方法中作的同樣。這就說明了變換流中的兩部分事實上被鏈接起來。

_flush() 方法只接受一個回調函數做爲參數，必須確保在全部操做完成以後調用它，使流終結。

through2 核心源碼解讀

through2.js 是 Transform stream 的簡單封裝庫，用起來很是簡單，下面來看下它的核心代碼。

function through2 (construct) {
  return function throughHOC (options, transform, flush) {
    if (typeof options == 'function') {
      flush     = transform
      transform = options
      options   = {}
    }

    if (typeof transform != 'function')
      transform = noop

    if (typeof flush != 'function')
      flush = null

    return construct(options, transform, flush)
  }
}
複製代碼

這是段工廠函數，through2.js 的三個 api 都是由這個方法生成的。

同時 through2 也是一個高階函數，接收一個參數，這個參數 construct 是一個函數，在這個項目中這個形參將有三個實參，也就是對應的三個 API。

through2 也返回一個高階函數，爲了能更好的認識這個函數，命名爲 throughHOC， 它有三個形式參數：

• options Transform 類的實例參數
• transform 實際轉換函數
• flush 方法只接受一個回調函數做爲參數，必須確保在全部操做完成以後調用它，使流終結

函數 throughHOC 內部對參數作了一些整理：

• 若是 options 是一個 function ，那這個 options 就是轉換函數，options 則會是一個默認值；
• 若是 options 存在且 transform 不是一個 function ，那 transform 就被重置爲默認轉換函數；
• 若是 flush 不是一個 function ，則重置爲 null

參數整理完了，就把它們做爲參數，傳入 construct 函數內。這個 construct 就是實現三個 API 的方法。

說 API 方法以前，先說下 Transform 類的加工 -- DestroyableTransform：

function DestroyableTransform(opts) {
  Transform.call(this, opts)
  this._destroyed = false
}

inherits(DestroyableTransform, Transform)

DestroyableTransform.prototype.destroy = function(err) {
  if (this._destroyed) return
  this._destroyed = true
  
  var self = this
  process.nextTick(function() {
    if (err)
      self.emit('error', err)
    self.emit('close')
  })
}
複製代碼

DestroyableTransform 繼承 Transform ，實現了 destroy 方法，當觸發了 destroy 後，須要手動觸發 close 事件。

下面就來講下實現三個 API 的函數：

主方法

let construct = function (options, transform, flush) {
  var t2 = new DestroyableTransform(options)

  t2._transform = transform

  if (flush)
    t2._flush = flush

  return t2
}

module.exports = through2(construct)
複製代碼

上面說過了 through2 函數，它的參數就是上面的 construct 函數，首先 實例化 DestroyableTransform 這個類，options 就是經過外部傳入的配置參數，接下來就是從新實現了 _transform 和 _flush 這兩個方法。

through2.obj

這個 API 和主方法惟一的區別就是開啓了對象模式，將 objectMode 屬性設置爲 true，highWaterMark 屬性設置爲 16

var t2 = new DestroyableTransform(Object.assign({ objectMode: true, highWaterMark: 16 }, options))
複製代碼

through2.ctor

這個 API 返回的是 DestroyableTransform 的子類，並非 Transform stream 的實例，這個在使用的時候其實和主方法惟一的區別就是須要額外實例化這個 API 返回值。

let construct = function (options, transform, flush) {
  function Through2 (override) {
    if (!(this instanceof Through2))
      return new Through2(override)

    this.options = Object.assign({}, options, override)

    DestroyableTransform.call(this, this.options)
  }

  inherits(Through2, DestroyableTransform)

  Through2.prototype._transform = transform

  if (flush)
    Through2.prototype._flush = flush

  return Through2
}

module.exports.ctor = through2(construct)
複製代碼

使用：

const through2 = require('through2')
const Ctor = through2.ctor(function(chunk, enc, callback) {
  console.log('chunk', chunk.toString());
  callback(null, chunk);
});
const th2 = new Ctor();
複製代碼

through2 使用 typescript 重構

不得不感嘆這個項目的厲害之處，僅僅只是對 Transform 作了一層簡單的封裝，卻透出了不少內容，項目中雖然只是額外擴展了兩個 api，可是熟悉源碼以後就能夠對它作更多的擴展了。這也不得不說轉換流 Transform 的強大之處。

在學習源碼以後，用 typescript 重構了一下，對代碼更加的清晰，有了更多的認識，值得好好學習。

源碼地址--through2-ts