Pular para o conteúdo

Fluxos

Ver como Markdown Editar no GitHub Documentação em 5b53f69

Phasis implementa o padrão WHATWG Streams completo. Todas as classes, todos os modos de leitura, todos os auxiliares. O mesmo código que funciona no Node e em navegadores modernos funciona em Phasis.

A implementação é a que fetch usa internamente — Response.body é um ReadableStream real, Request aceita corpos em fluxo, e o tee() necessário para clonar uma Response funciona.

O que foi implementado

Classe Status
ReadableStream Completa
ReadableStreamDefaultController Completa
ReadableByteStreamController Completa
ReadableStreamDefaultReader Completa
ReadableStreamBYOBReader Completa
ReadableStreamBYOBRequest Completa
WritableStream Completa
WritableStreamDefaultController Completa
WritableStreamDefaultWriter Completa
TransformStream Completa
TransformStreamDefaultController Completa
ByteLengthQueuingStrategy Completa
CountQueuingStrategy Completa

Além dos métodos que você costuma usar: pipeTo, pipeThrough, tee, getReader, iteração assíncrona, Symbol.asyncIterator, leitura BYOB com { min }.

Produção

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();
  },
});

Consumo

Três formas de ler:

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)]);

Transformação

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

Fluxos de bytes (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 } é compatível — read(view, { min: 2 }) só é resolvido quando há pelo menos 2 elementos disponíveis.

Contrapressão

desiredSize e ready funcionam conforme descrito na especificação. Use uma estratégia de enfileiramento para fornecer um canal de retorno ao produtor:

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

Com 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 para armazenamento em cache

Padrão comum: ler o corpo duas vezes.

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

Response.clone() faz isso por você e retorna um Response novo.

Observações

  • Produtores apenas síncronos. Um pull() que executa E/S PHP bloqueante bloqueará o script. Use o hook de transporte fetch para HTTP; para arquivos, prefira lê-los na memória previamente. setTimeout / setInterval estão disponíveis (consulte Loop de eventos), mas o mecanismo usa tempo virtual por padrão — um setTimeout(cb, 1000) dentro de um pull() não dorme PHP, ele avança o relógio do loop. Para cadência em tempo real, o host precisa executar tickEventLoop() em relação a um relógio de parede.

Veja também