Streams API — ReadableStream, WritableStream으로 데이터 스트리밍

// 커스텀 ReadableStream 생성
const stream = new ReadableStream({
  start(controller) {
    // 초기화
    controller.enqueue("첫 번째 청크");
    controller.enqueue("두 번째 청크");
    controller.close(); // 스트림 종료
  },
});

// 읽기
const reader = stream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(value);
}

function createCounterStream(limit) {
  let count = 0;
  return new ReadableStream({
    pull(controller) {
      if (count >= limit) {
        controller.close();
        return;
      }
      // pull은 소비자가 데이터를 요청할 때 호출됨
      controller.enqueue(count++);
    },
  });
}

const stream = createCounterStream(5);
const reader = stream.getReader();

let result;
while (!(result = await reader.read()).done) {
  console.log(result.value); // 0, 1, 2, 3, 4
}

const writableStream = new WritableStream({
  write(chunk) {
    console.log("쓰기:", chunk);
  },
  close() {
    console.log("스트림 닫힘");
  },
  abort(reason) {
    console.error("중단:", reason);
  },
});

const writer = writableStream.getWriter();
await writer.write("데이터 1");
await writer.write("데이터 2");
await writer.close();

// 대문자 변환 스트림
const upperCaseTransform = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

// 파이프라인: 읽기 → 변환 → 쓰기
const readable = new ReadableStream({
  start(controller) {
    controller.enqueue("hello");
    controller.enqueue("world");
    controller.close();
  },
});

const results = [];
await readable
  .pipeThrough(upperCaseTransform)
  .pipeTo(new WritableStream({
    write(chunk) { results.push(chunk); },
  }));

console.log(results); // ["HELLO", "WORLD"]

// 대용량 파일 다운로드 + 진행률
async function downloadWithProgress(url) {
  const response = await fetch(url);
  const total = parseInt(response.headers.get("Content-Length") || "0");
  let loaded = 0;

  const reader = response.body.getReader();
  const chunks = [];

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

    chunks.push(value);
    loaded += value.length;

    if (total > 0) {
      console.log(`진행률: ${((loaded / total) * 100).toFixed(1)}%`);
    }
  }

  return new Blob(chunks);
}

// 서버에서 줄 단위 JSON을 스트리밍으로 받기
async function* parseNDJSON(response) {
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

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

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop(); // 미완성 줄은 버퍼에 유지

    for (const line of lines) {
      if (line.trim()) {
        yield JSON.parse(line);
      }
    }
  }

  if (buffer.trim()) {
    yield JSON.parse(buffer);
  }
}

// 사용
const response = await fetch("/api/stream");
for await (const item of parseNDJSON(response)) {
  console.log(item); // 각 JSON 객체가 도착할 때마다 처리
}

// 여러 변환을 체이닝
const compressionStream = new CompressionStream("gzip");
const decompressionStream = new DecompressionStream("gzip");

// 텍스트 → 압축 → 전송
const blob = new Blob(["대용량 텍스트 데이터..."]);
const compressed = blob.stream()
  .pipeThrough(new CompressionStream("gzip"));

// 서버로 압축 전송
await fetch("/api/upload", {
  method: "POST",
  body: compressed,
  headers: { "Content-Encoding": "gzip" },
});

// 하나의 스트림을 두 갈래로 분기
const response = await fetch("/api/data");
const [stream1, stream2] = response.body.tee();

// stream1은 화면에 표시
displayStream(stream1);

// stream2는 캐시에 저장
cacheStream(stream2);