Node Stream pipe的誕生

流在Node.js中是處理流數據的抽象接口。stream模塊提供了基礎的API。使用這些API能夠很容易地來構建實現流接口的對象。 Node.js裏提供了不少流對象。例如http.IncomingMessage類、fs.createReadStream類等等。都繼承流的私有屬性和公有方法。 因此學習流，有助於學習Node的其餘模塊。

文章結構

• 簡介stream
• pipe源碼

stream

const EE = require('events');
const util = require('util');
function Stream() {
  EE.call(this);
}
util.inherits(Stream, EE);
複製代碼

• Stream繼承EventEmitter。流能夠是可讀的、可寫的，或是可讀寫的。
• Stream分爲Readable（可讀流）、Writable（可寫流）、Duplex（可讀寫流）、Transform（讀寫過程當中能夠修改和變換數據的 Duplex 流）。

pipe

Stream.prototype.pipe = function(dest, options){
    var source = this;
    source.on('data', ondata);
    dest.on('drain', ondrain);
    if (!dest._isStdio && (!options || options.end !== false)) {
        source.on('end', onend);
        source.on('close', onclose);
    }
    source.on('error', onerror);
    dest.on('error', onerror);
    source.on('end', cleanup);
    source.on('close', cleanup);  
    dest.on('close', cleanup);
    dest.emit('pipe', source);
    return dest;
};
複製代碼

• Stream公有方法pipe
  • source是可讀流：Readable。dest是可寫流：Writable。
  • Readable.pipe(Writable) 。
  • Readable訂閱事件：data、error、end、close。Readable接收到事件執行相應的方法。
  • Writable訂閱事件：drain、error、close，併發布pipe事件。Writable接收到事件執行相應的方法。
  • 返回Writable。

ondata

function ondata(chunk) {
    if (dest.Writable) {
      if (false === dest.write(chunk) && source.pause) {
        source.pause();
      }
    }
  }
複製代碼

• Readable訂閱data事件。
• Readable觸發data事件，表示讀入數據。
• dest.Writable當寫完時會賦值爲false。
• 若是讀的太快，沒有寫完dest.write(chunk)返回false。
• source.pause暫停寫入。
• 總結：訂閱data事件，觸發ondata方法，若是Readable讀入數據太快，來不及寫入，要暫停讀入數據。

ondrain

function ondrain() {
    if (source.Readable && source.resume) {
      source.resume();
    }
  }
複製代碼

• Writable訂閱drain事件。
• Writable觸發drain事件，表示這時才能夠繼續向流中寫入數據。
• source.Readable在讀到末尾時會賦值爲false。
• source.resume()表示會從新觸發Writable的data事件。
• 總結：訂閱drain事件，表示這時才能夠繼續向流中寫入數據，調用source.resume()，觸發Writable的data事件。

cleanup

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);
  }

複製代碼

• 移除Writable和Readable訂閱的事件。

error

function onerror(er) {
    cleanup();
    if (EE.listenerCount(this, 'error') === 0) {
      throw er; // Unhandled stream error in pipe.
    }
  }
複製代碼

• Writable和Readable有錯誤是執行的方法。

options

if (!dest._isStdio && (!options || options.end !== false)) {
    source.on('end', onend);
    source.on('close', onclose);
}
複製代碼

• dest._isStdio 暫時沒理解。
• 若是不傳options或者options.end不是false，給Readable訂閱end和close事件。

onend

var didOnEnd = false;
  function onend() {
    if (didOnEnd) return;
    didOnEnd = true;

    dest.end();
  }
複製代碼

• 當Readable觸發end事件時，執行dest.end()，中止寫入。

onclose

function onclose() {
    if (didOnEnd) return;
    didOnEnd = true;

    if (typeof dest.destroy === 'function') dest.destroy();
  }

複製代碼

• 當Readable觸發close事件後，該流將不會再觸發任何事件。
• dest.destroy()摧毀這個流，併發出傳過來的錯誤。當這個函數被調用後，這個寫入流就結束了。

小記

• pipe方法大白話：讀東西，寫東西，讀快了，來不及寫，暫停讀，來得及寫了，再讀東西，再寫。。。
• 流在Node.js中應用普遍，http、文件、打包工具等等。可見流的重要性。
• 學習源碼，有助於理解底層的實現。

參考

• Stream
• pipe