스트림

Markdown으로 보기 GitHub에서 수정 문서 위치 5b53f69

Phasis는 WHATWG Streams Standard 전체를 구현합니다. 모든 클래스, 모든 리더 모드, 모든 헬퍼를 제공합니다. Node와 최신 브라우저에서 작동하는 코드가 Phasis에서도 작동합니다.

구현된 기능은 fetch 내부에서 사용하는 것과 같습니다. Response.body는 실제 ReadableStream이며 Request는 스트리밍 본문을 지원하고, Response를 복제하는 데 필요한 tee()도 작동합니다.

제공 기능

클래스	상태
`ReadableStream`	전체
`ReadableStreamDefaultController`	전체
`ReadableByteStreamController`	전체
`ReadableStreamDefaultReader`	전체
`ReadableStreamBYOBReader`	전체
`ReadableStreamBYOBRequest`	전체
`WritableStream`	전체
`WritableStreamDefaultController`	전체
`WritableStreamDefaultWriter`	전체
`TransformStream`	전체
`TransformStreamDefaultController`	전체
`ByteLengthQueuingStrategy`	전체
`CountQueuingStrategy`	전체

자주 사용하는 메서드도 지원합니다. pipeTo, pipeThrough, tee, getReader, 비동기 반복, Symbol.asyncIterator, { min }를 사용한 BYOB 읽기가 가능합니다.

생산

JavaScript

// Pull-based — only generates values when read
const counter = new ReadableStream({
  start(controller) { this.n = 0; },
  pull(controller) {
    if (this.n >= 5) { controller.close(); return; }
    controller.enqueue(this.n++);
  },
});

for await (const v of counter) console.log(v);  // 0,1,2,3,4

JavaScript

// Push-based — start enqueues everything eagerly
const small = new ReadableStream({
  start(c) {
    [10, 20, 30].forEach(n => c.enqueue(n));
    c.close();
  },
});

소비

읽는 방법은 세 가지입니다.

JavaScript

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

// 2. Async iteration (preferred)
for await (const chunk of stream) {
  console.log(chunk);
}

// 3. tee + parallel pipes
const [a, b] = stream.tee();
await Promise.all([a.pipeTo(sinkA), b.pipeTo(sinkB)]);

변환

JavaScript

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

const result = new ReadableStream({
  start(c) { ['hi', 'there'].forEach(s => c.enqueue(s)); c.close(); }
}).pipeThrough(upper);

for await (const v of result) console.log(v);  // HI, THERE

바이트 스트림(BYOB)

JavaScript

const bytes = new ReadableStream({
  type: 'bytes',
  start(controller) {
    controller.enqueue(new Uint8Array([1, 2, 3, 4, 5, 6]));
    controller.close();
  },
});

const reader = bytes.getReader({ mode: 'byob' });
const buf = new Uint8Array(4);
const { value, done } = await reader.read(buf);
// value is a Uint8Array of length 4, sharing buffer with `buf`

{ min: N }를 지원합니다. read(view, { min: 2 })는 최소 2개 요소를 사용할 수 있을 때만 확인됩니다.

백프레셔

desiredSize와 ready는 사양에 설명된 대로 작동합니다. 큐잉 전략을 사용해 생산자에게 역방향 채널을 제공하세요.

JavaScript

const stream = new ReadableStream({
  pull(controller) {
    if (controller.desiredSize <= 0) return;  // back off
    controller.enqueue(nextChunk());
  },
}, new CountQueuingStrategy({ highWaterMark: 4 }));

fetch와 함께 사용

JavaScript

// Response body is a ReadableStream
const res = await fetch('https://example.com/big.txt');
let total = 0;
for await (const chunk of res.body) {
  total += chunk.byteLength;
}

// Request bodies can stream too
const upload = new ReadableStream({
  start(c) { c.enqueue(new TextEncoder().encode('hello')); c.close(); }
});
await fetch('https://example.com/upload', { method: 'POST', body: upload });

캐싱을 위한 tee

일반적인 패턴: 본문을 두 번 읽습니다.

JavaScript

const res = await fetch(url);
const [forCache, forUse] = res.body.tee();
await Promise.all([
  forCache.pipeTo(cacheWritable),
  parseJson(forUse),
]);

Response.clone()가 이를 처리하고 새로운 Response를 반환합니다.

주의 사항

동기 방식 생산자만 지원합니다. 블로킹 PHP I/O를 수행하는 pull()는 스크립트를 멈춥니다. HTTP에는 fetch 전송 훅을 사용하고, 파일은 미리 메모리로 읽는 것이 좋습니다. setTimeout / setInterval도 사용할 수 있습니다(이벤트 루프 참조). 다만 엔진은 기본적으로 가상 시간을 사용하므로 pull() 안의 setTimeout(cb, 1000)는 PHP 동안 대기하는 대신 루프 시계를 진행시킵니다. 실시간 간격 조절을 위해서는 호스트가 벽시계에 맞춰 tickEventLoop()를 구동해야 합니다.

스트림

제공 기능

생산

소비

변환

바이트 스트림(BYOB)

백프레셔

fetch와 함께 사용

캐싱을 위한 tee

주의 사항

관련 항목