Stream.pipeline() Method

nodejs에서 스트림은 핵심 모듈부터 시작하여 Node.js의 모든 곳에 존재합니다. 예를 들어, fs 모듈에는 파일 읽기용 createReadStream() 및 파일 쓰기용 createWriteStream()이 있으며, HTTP요청 및 응답 객체는 기본적으로 스트림이며, zlib 모듈을 사용하면 스트림 인터페이스를 사용하여 데이터를 압축 및 압축 해제할 수 있습니다. 또한 암호화 모듈조차도 createCipheriv와 createDecipheriv와 같은 몇몇 유용한 스트리밍 기본 요소들을 노출시키고 있습니다.

Node.js의 모든 스트림은 스트림 코어 모듈에서 사용할 수 있는 네 가지 기본 추상 클래스 중 하나의 구현입니다.

• Readable
• Writable
• Duplex
• Transform

각 스트림 클래스는 EventEmitter의 인스턴스이기도 합니다. 실제로 스트림은 읽기 가능한 스트림이 읽기를 마쳤을 때 'end', 쓰기 스트림이 쓰기를 완료했을 때, 'finish', 무언가 잘못되었을 때 'error'와 같은 여러 유형의 이벤트를 생성할 수 있습니다.

스트림이 매우 유연한 이유 중 하나는 바이너리 데이터뿐만 아니라 거의 모든 JavaScript의 값을 처리할 수 있다는 사실입니다.

• Binary 모드: 버퍼 또는 문자열과 같은 청크 형태로 데이터를 스트리밍합니다.
• 객체모드: 데이터를 일련의 개별 객체로 스트리밍합니다. (거의 모든 JavaScript 값을 사용할 수 있음.)

이 두 가지 작동 모드를 통해 I/O뿐만 아니라 함수 방식으로 처리 단위를 우아하게 구성할 수 있는 도구로도 사용할 수 있습니다.

다음은 pipeline 함수의 메소드를 살펴보겠습니다.

The stream.pipeline() method is a module method that is used to the pipe by linking the streams on errors and accurately cleaning up and providing a callback function when the pipeline is done.

Syntax: stream.pipeline(...streams, callback)

@param

This.method accepts two paramters as mentioned above and described below.

• ...streams: These are two or more streams which are to be piped together.
• callback: This function is called when the pipeline is fully done and it shows an 'error' if the pipeline is not accomplished.

@returns

It returns a cleanup function.

Below examples illustrate the use of stream.pipeline() method in Node.js

Example 1:

// 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 {
    await pipelineAsync(readable, transform, writable);
    console.log('pipeline accomplished.');
  } catch(err) {
    console.error('pipeline failed with error', err);
  }
})();

Output:

Promise{}
pipeline accomplished.

Example 2:

var fs = require('fs');
const zlib = require('zlib');

const { pipeline } = require('stream');
const { promisify } = require('util');
const pipelineAsync = promisify(pipeline);

const readable = fs.createReadStream('input.text');
const writable = fs.createWriteStream('output.text');
const transform = zlib.createGzip();


(async function run() {
  try {
    await pipelineAsync(readable, writable, transform);
    console.log('pipeline accomplished.');
  } catch(err) {
    console.error('pipeline failed with error:', err);
  }
})();

Output

Promise { }
pipeline failed with error: Error [ERR_STREAM_CANNOT_PIPE]: Cannot pipe, not readable at WriteStream.Writable.pipe