Node.js Stream.pipeline() 方法

stream.pipeline() 方法是用于管道的模块方法，通过链接传递错误的流并在管道完成时准确清理并提供回调函数。

句法：

stream.pipeline(...streams, callback)

参数：此方法接受上面提到的和下面描述的两个参数。

• …流：这是两个或多个要通过管道连接在一起的流。
• 回调：当管道完全完成时调用此函数，如果管道未完成，则显示“错误”。

返回值：它返回一个清理函数。

下面的例子说明了 Node.js 中stream.pipeline() 方法的使用：

示例 1：

// Node.js program to demonstrate the     
// stream.pipeline() method
  
// Including fs and zlib module
var fs = require('fs');
const zlib = require('zlib');
  
// Constructing finished from stream
const { pipeline } = require('stream');
  
// Constructing promisify from
// util
const { promisify } = require('util');
  
// Defining pipelineAsync method
const pipelineAsync = promisify(pipeline);
  
// Constructing readable stream
const readable = fs.createReadStream("input.text");
  
// Constructing writable stream
var writable = fs.createWriteStream("output.text");
  
// Creating transform stream
const transform = zlib.createGzip();
  
// Async function
(async function run() {
  try{ 
  
    // pipelining three streams
    await pipelineAsync(
      readable,
      transform,
      writable
    );
    console.log("pipeline accomplished.");
    }
  
  // Shows error
  catch(err) {
    console.error('pipeline failed with error:', err);
  }
  })();

输出：

Promise {  }
pipeline accomplished.

示例 2：

// Node.js program to demonstrate the     
// stream.pipeline() method
  
// Including fs and zlib module
var fs = require('fs');
const zlib = require('zlib');
  
// Constructing finished from stream
const { pipeline } = require('stream');
  
// Constructing promisify from
// util
const { promisify } = require('util');
  
// Defining pipelineAsync method
const pipelineAsync = promisify(pipeline);
  
// Constructing readable stream
const readable = fs.createReadStream("input.text");
  
// Constructing writable stream
var writable = fs.createWriteStream("output.text");
  
// Creating transform stream
const transform = zlib.createGzip();
  
// Async function
(async function run() {
  try{ 
  
    // pipelining three streams
    await pipelineAsync(
      readable,
      writable,
      transform
    );
    console.log("pipeline accomplished.");
    }
  
  // Shows error
  catch(err) {
    console.error('pipeline failed with error:', err);
  }
  })();

输出：这里，管道时流的顺序不正确，因此发生错误。

Promise {  }
pipeline failed with error: Error [ERR_STREAM_CANNOT_PIPE]: Cannot pipe, not readable
    at WriteStream.Writable.pipe (_stream_writable.js:243:24)
    at pipe (internal/streams/pipeline.js:57:15)
    at Array.reduce ()
    at pipeline (internal/streams/pipeline.js:88:18)
    at Promise (internal/util.js:274:30)
    at new Promise ()
    at pipeline (internal/util.js:273:12)
    at run (/home/runner/ThirstyTimelyKey/index.js:33:11)
    at /home/runner/ThirstyTimelyKey/index.js:45:5
    at Script.runInContext (vm.js:133:20)

参考： https://nodejs.org/api/stream.html#stream_stream_pipeline_streams_callback。