有道unpipe一個streams2管道和清空（不只是沖水）

Question

前提

我試圖找到正確的方式提前終止了一系列的Node.js管道流（管道）的：有時我要在流完成之前正常中止流。具體而言，我主要處理objectMode: true和非本地平行流，但這應該不重要。

問題

問題是，當我unpipe管道，數據保持在每個流的緩衝器，並且drain編對於大多數中間流（例如，Readable/Transform），這個可能是，但最後的Writable仍然排到其寫入目標（例如文件或數據庫或套接字或w/e）。如果緩衝區包含數百或數千個需要耗費大量時間的塊，這可能會產生問題。我希望它立即停止，即不排水;爲什麼浪費週期和內存對數據無關緊要？

根據我走的路線，我收到「寫完後」錯誤，或者當流找不到現有管道時發生異常。

問題

什麼是有道優雅地殺死流的形式a.pipe(b).pipe(c).pipe(z)管道？

解決方案？

我已經提出瞭解決的辦法是3步：

1. unpipe管道中的每個流以相反的順序
2. 實現Writable
3. end每個實現Writable流空每個流的緩衝器

說明整個過程的一些僞代碼：

var pipeline = [ // define the pipeline 
    readStream, 
    transformStream0, 
    transformStream1, 
    writeStream 
]; 

// build and start the pipeline 
var tmpBuildStream; 
pipeline.forEach(function(stream) { 
    if (!tmpBuildStream) { 
     tmpBuildStream = stream; 
     continue; 
    } 
    tmpBuildStream = lastStream.pipe(stream); 
}); 

// sleep, timeout, event, etc... 

// tear down the pipeline 
var tmpTearStream; 
pipeline.slice(0).reverse().forEach(function(stream) { 
    if (!tmpTearStream) { 
     tmpTearStream = stream; 
     continue; 
    } 
    tmpTearStream = stream.unpipe(tmpTearStream); 
}); 

// empty and end the pipeline 
pipeline.forEach(function(stream) { 
    if (typeof stream._writableState === 'object') { // empty 
    stream._writableState.length -= stream._writableState.buffer.length; 
    stream._writableState.buffer = []; 
    } 
    if (typeof stream.end === 'function') { // kill 
    stream.end(); 
    } 
}); 

我真的很擔心的stream._writableState使用和修改內部buffer和length特性（_意味着私有財產）。這看起來像一個黑客。還請注意，由於我是管道系統，因此我們無法解決問題（基於我從IRC收到的建議），如pause和resume。

我也把一個可運行的版本（相當草率），你可以從GitHub抓取：https://github.com/zamnuts/multipipe-proto（GIT克隆，NPM安裝，查看自述，NPM開始）

Answer 1

在這種特殊情況下，我覺得我們應該得到擺脫你有4個不同的不完全定製的流的結構。如果我們沒有實現我們自己的機制，將它們連接在一起將產生鏈條依賴性，這將很難控制。

我想專注於自己的實際工作中的目標位置：

INPUT >----[read] → [transform0] → [transform1] → [write]-----> OUTPUT 
       |   |    |   | 
KILL_ALL------o----------o--------------o------------o--------[nothing to drain] 

我認爲，上述結構可以通過組合定製來實現：

1. duplex stream - 爲自己的_write(chunk, encoding, cb)和_read(bytes)用

   執行

2. transform stream - 爲自己的_transform(chunk, encoding, cb)執行。

由於您使用的是writable-stream-parallel包您可能還需要去在他們的庫，因爲他們的duplex實現可以在這裏找到：https://github.com/Clever/writable-stream-parallel/blob/master/lib/duplex.js。 而他們的transform stream實現在這裏：https://github.com/Clever/writable-stream-parallel/blob/master/lib/transform.js。他們在這裏處理highWaterMark。

可能的解決方法

他們寫流：https://github.com/Clever/writable-stream-parallel/blob/master/lib/writable.js#L189有一個有趣的功能writeOrBuffer，我想你也許可以調整它有點中斷從緩存寫入數據。

注意：這些3 flags所控制的緩衝器清除：

(!finished && !state.bufferProcessing && state.buffer.length) 

---

參考文獻：

• Node.js Transform Stream Doc
• Node.js Duplex Stream Doc
• Writing Transform Stream in Node.js
• Writing Duplex Stream in Node.js