내장 fetch와 Web Streams — 스트리밍 깊게 보기

"트랙 5에서는 fetch의 기본을 봤어. 운영 코드에서는 짧은 예제 뒤가 더 중요해. 요청 본문을 흘려보내고, 잘린 조각을 이어 붙이고, Web Stream과 Node 스트림을 연결하며, 언제 낮은 수준의 undici가 필요한지 판단해야 해."

큰 본문은 스트림으로 올려

작은 POST 요청에는 body: JSON.stringify(...)면 충분해. 하지만 큰 영상이나 계속 쌓이는 로그를 보낼 때는 본문 전체를 메모리에 올리지 말고 조각마다 흘려보내야 해.

import { createReadStream } from 'node:fs';
import { Readable } from 'node:stream';

const src = createReadStream('./huge.bin');

const res = await fetch('https://uploader.example.com/upload', {
  method: 'PUT',
  body: Readable.toWeb(src),
  duplex: 'half',
  headers: { 'Content-Type': 'application/octet-stream' },
});
console.log('status:', res.status);

Node에서 스트림을 fetch 요청 본문으로 보낼 때는 duplex: 'half'를 적어야 해. 스트리밍 본문을 보낸다는 사실을 명시하는 옵션이라서, 빠뜨리면 Node가 요청을 보내기 전에 오류를 내. 이름에 half가 들어가도 파일 전체를 먼저 버퍼링한다는 뜻은 아니야. ReadableStream이 내놓는 조각이 차례로 요청에 쓰여.

SSE는 오래 이어지는 HTTP 응답이야

Server-Sent Events는 Content-Type: text/event-stream인 HTTP 응답이야. fetch는 그 본문을 Web Stream으로 내줘. 다만 네트워크 조각이 줄 끝에서 정확히 잘린다는 보장은 없으니, 읽다 만 꼬리를 다음 조각과 이어야 해.

const res = await fetch('https://api.example.com/stream', {
  headers: { Accept: 'text/event-stream' },
});
if (!res.ok || !res.body) throw new Error(`SSE failed: ${res.status}`);

const reader = res.body
  .pipeThrough(new TextDecoderStream())
  .getReader();

let pending = '';
while (true) {
  const { value, done } = await reader.read();
  if (done) break;

  pending += value;
  const lines = pending.split('\n');
  pending = lines.pop() ?? '';

  for (const line of lines) {
    if (line.startsWith('data: ')) {
      handleEvent(JSON.parse(line.slice(6)));
    }
  }
}

이 코드는 한 줄짜리 data:마다 JSON 하나를 보내는 단순한 스트림에 맞고, 한 줄이 두 조각으로 나뉘어 와도 제대로 이어. 완전한 SSE 규격에는 여러 줄 데이터, 이벤트 이름, ID, 재시도 시간, 주석, CRLF 구분도 있어. 서버가 그 기능을 쓴다면 검증된 완전한 파서를 쓰는 편이 안전해.

세밀한 제어가 필요할 때만 undici를 직접 써

import { Pool } from 'undici';

const pool = new Pool('https://api.example.com', {
  connections: 100,
  pipelining: 10,
  bodyTimeout: 30_000,
});

const { body, statusCode } = await pool.request({
  method: 'GET',
  path: '/items',
});

for await (const chunk of body) {
  // chunk is a Buffer
}

Response로 스트림을 감쌀 수 있어

Response는 fetch가 돌려주는 값으로만 쓰는 객체가 아니야. 변환한 스트림을 감싸 새 HTTP 응답을 만들 수도 있어서 프록시, 캐시, 서버 처리 함수에 유용해.

const upstream = await fetch('https://api.example.com/big.json');

const upper = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk);
  },
});

const piped = upstream.body.pipeThrough(upper);
const response = new Response(piped, {
  status: upstream.status,
  headers: upstream.headers,
});

// `response` can be returned by a compatible server handler,
// or consumed with .text(), .json(), and the other Body methods.

Pippa의 고백