Readable 와 Writable — 두 기초

"Readable 은 청크 생산. Writable 은 청크 소비. 다른 모든 stream 타입은 이 둘의 리믹스야."

Readable — 데이터 오는 곳

Readable stream 은 청크의 sequence 생산. 써본 예: fs.createReadStream('path'), process.stdin, http.IncomingMessage (서버에서 req 객체). 생산자가 바이트 (또는 object mode 면 객체) push; 소비자가 읽음.

모던 인터페이스 — async iteration:

import { createReadStream } from 'node:fs';

const stream = createReadStream('input.txt', { encoding: 'utf-8' });
for await (const chunk of stream) {
  console.log('got', chunk.length, 'chars');
}

레거시 이벤트 인터페이스 (callback 에서 여전히 흔함):

stream.on('data', chunk => { /* one chunk at a time */ });
stream.on('end',  ()    => { /* no more chunks */ });
stream.on('error', err => { /* handle */ });

Writable — 데이터 가는 곳

Writable stream 은 청크 소비. 예: fs.createWriteStream('path'), process.stdout, http.ServerResponse (res 객체). 바이트 넣음; 목적지가 write.

import { createWriteStream } from 'node:fs';

const out = createWriteStream('output.txt');
out.write('hello\n');
out.write('world\n');
out.end();   // flush and close

out.write(chunk) 가 internal buffer 에 자리 있으면 true 반환, 목적지가 drain 할 수 있는 속도보다 빠르게 쓰고 있으면 false. 그 false 반환이 backpressure 신호 — 'drain' 들을 때까지 쓰기 멈춰야 함.

Object Mode vs Binary Mode

Stream 기본은 바이트. Object mode ({ objectMode: true }) 가 임의의 JavaScript 값을 청크로 넘기게 해 — record, 로그 엔트리, 파싱된 이벤트 streaming 할 때 유용. Object-mode stream 이 살짝 느려 (internal pooling 없음), 청크가 바이트 아닐 때 편의가 거대. 예:

CSV 파서 stream: 입력 바이트, 출력 객체.
EventEmitter 를 이벤트 stream 으로 감쌈.
DB cursor 를 row stream 으로 감쌈.

한 stream 에서 모드 섞기 금지 — 생성 시 선언.

Flowing vs Paused 모드

Readable stream 은 두 모드:

Paused — 데이터가 internal buffer 에 앉아 있다가 .read() 로 요청해야 옴. 기본.
Flowing — 데이터가 source 가 제공하는 속도로 너한테 push. 'data' listener 붙이거나 .resume() 호출로 trigger.

for await...of 가 투명하게 처리 — 한 번에 한 청크 요청, pause, 또 요청. async iteration 쓰면 모드 신경 안 써도 됨. flowing/paused 구분은 레거시 이벤트 API 와 수동 flow control 섞을 때 주로 중요해.

네 Readable 만들기

subclass 할 일 드문데, 명시적 방법:

import { Readable } from 'node:stream';

class Counter extends Readable {
  constructor(max) {
    super({ objectMode: true });
    this.i = 0;
    this.max = max;
  }
  _read() {
    if (this.i >= this.max) this.push(null);    // null = end
    else this.push({ n: this.i++ });
  }
}

for await (const obj of new Counter(5)) {
  console.log(obj);   // { n: 0 }, { n: 1 }, ...
}

대부분 경우 Readable.from(iterableOrAsyncIterable) 가 더 단순하고 subclass 완전 피함.

Pippa 의 고백

오랫동안 난 Node stream 이 복잡한 줄 알았어, 모든 튜토리얼이 .on('data'), .on('end'), .pause()/.resume() 춤 보여줬으니까. 아빠가 짚어줬어: "그게 레거시 API. 모던 인터페이스는 `for await (const chunk of readable)` — 같은 머신, ergonomic 한 표면." async-iterator 렌즈로 docs 읽으니 stream 이 baroque 안 느껴졌어. 이제 크거나 unbounded 한 입력엔 본능적으로 stream 손대, 코드가 단순 `readFile` 버전이랑 거의 비슷해 — 망가질 방법 하나만 빠진.

Code

Readable.from() — stream 만드는 저렴한 방법·javascript

// Readable from any iterable — the easy way
import { Readable } from 'node:stream';

// From an array
const nums = Readable.from([1, 2, 3, 4, 5], { objectMode: true });
for await (const n of nums) console.log(n);

// From an async generator
async function* paginated() {
  for (let p = 1; p <= 10; p++) {
    const data = await fetchPage(p);
    yield* data;   // yield each item separately
  }
}
const stream = Readable.from(paginated(), { objectMode: true });

수동 backpressure — pipeline 이 너 대신 해주는 것·javascript

// Backpressure-aware writing
import { createWriteStream } from 'node:fs';

const out = createWriteStream('out.txt');

async function pumpLots(source) {
  for await (const chunk of source) {
    if (!out.write(chunk)) {
      // Internal buffer full — wait for drain
      await new Promise(r => out.once('drain', r));
    }
  }
  out.end();
}

// Better: just use pipeline (next lesson) — it handles backpressure for you
// import { pipeline } from 'node:stream/promises';
// await pipeline(source, out);

Exercise

파일에서 줄 Readable stream 반환하는 lineStream(path) 함수 짜 (한 청크당 한 줄, 끝의 newline 없이). 필터 + 카운트에 써: 'ERROR' 포함하는 줄만 출력하고 총 count 추적하는 함수에 pipe. readline 직접 쓰는 거랑 비교. 언제 네 커스텀 stream 을 readline 보다 선호할까?

Hint

createReadStream(path) 둘레에 createInterface 감싸고, 그 다음 Readable.from(rl, { objectMode: true }). 그게 더 큰 pipeline 으로 pipe 할 수 있는 라인 객체 Readable 줘. 커스텀 stream 가치는 더 큰 pipeline 으로 compose 하고 싶을 때 — readline 혼자선 다른 Node stream 으로 native 하게 compose 안 됨.