스트림(Stream)과 파이프라인

스트림(Stream)과 파이프라인

스트림은 데이터를 한 번에 다 받지 않고 조각조각 처리하게 해 주는 추상 인터페이스예요. HTTP 요청 객체나 process.stdout이 모두 스트림 인스턴스지요. 파일 전체를 메모리에 올리지 않고 큰 데이터를 흘려보내면서 다루고 싶을 때 특히 유용해요. 이 글에서는 스트림의 네 가지 기본 유형과, 파이프라인으로 여러 스트림을 이어 쓰는 법을 정리해 볼게요.

출처: Stream — Node.js 공식 문서

네 가지 스트림 유형

node:stream 모듈에는 네 가지 기본 스트림 유형이 있어요.

  • Writable: 데이터를 쓸 수 있는 스트림(예: fs.createWriteStream()).
  • Readable: 데이터를 읽을 수 있는 스트림(예: fs.createReadStream()).
  • Duplex: 읽고 쓰기를 모두 할 수 있는 스트림(예: net.Socket).
  • Transform: 읽고 쓰는 동안 데이터를 가공·변환하는 Duplex(예: zlib.createDeflate()).

모든 스트림은 EventEmitter의 인스턴스라서 이벤트를 기반으로 동작해요. 데이터를 소비하는 일이 목적이라면 보통 node:stream 모듈을 직접 쓸 필요는 없어요. 이미 주어진 스트림을 쓰면 되거든요.

파이프라인으로 이어 붙이기

여러 스트림을 순서대로 연결해서 에러 처리까지 한 번에 하고 싶다면 pipeline을 써요. 콜백 방식과 Promise 방식(stream/promises)이 모두 있어요.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);

pipeline에 AbortSignal을 options로 넘기면, 시그널이 중단됐을 때 파이프라인에 AbortError와 함께 destroy가 호출돼요. async generator도 파이프라인에 참여할 수 있고, 이때 source에 signal 인자가 전달되니 꼭 처리해 줘야 파이프라인이 영원히 끝나지 않는 문제를 피할 수 있어요.

스트림이 끝났는지 확인하기

finished()는 스트림이 더 이상 읽거나 쓸 수 없게 됐을 때 fulfilled되는 Promise를 반환해요. 파일을 다 읽었는지 확인하고 싶을 때 유용하죠.

const { finished } = require('node:stream/promises');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // Drain the stream.

주의할 점은, finished()가 '이 스트림의 읽기가 실제로 소비됐는지'까지는 보장하지 않는다는 거예요. 위처럼 rs.resume()으로 스트림이 흐르도록 해 줘야 finished Promise가 끝나요.

읽기 흐름 제어: pause, resume, pipe

Readable 스트림은 flowing 모드와 paused 모드가 있어요. 'data' 리스너를 달아서 데이터가 흘러나가는 상태를 flowing 모드라고 하고요. pause()를 호출하면 'data' 이벤트 전송을 멈추고, 그동안 쌓인 데이터는 내부 버퍼에 남아요. 다시 resume()으로 흐르게 할 수 있어요.

pipe(destination)는 읽는 스트림의 데이터를 쓰는 스트림으로 연결해 주는 편리한 방법이에요. 소스가 끝나면 end 옵션(기본 true)에 따라 목적지 스트림도 함께 끝나요.

한 가지 알아둘 점. 'readable' 이벤트와 'data' 이벤트를 동시에 쓰면 'readable'이 흐름 제어의 우선권을 가져요. 그러면 'data'는 stream.read()를 호출할 때만 발생하게 되고, readableFlowing은 false가 돼요.

더 알아보기