Lewati ke konten

Streams

Lihat sebagai Markdown Edit di GitHub Dokumentasi di 5b53f69

Phasis mengimplementasikan Standar WHATWG Streams secara lengkap. Semua kelas, semua mode pembaca, semua helper. Kode yang sama dan berfungsi di Node serta browser modern juga berfungsi di Phasis.

Implementasi ini adalah yang digunakan fetch secara internal — Response.body adalah ReadableStream sungguhan, Request menerima body streaming, dan tee() yang Anda perlukan untuk mengkloning Response berfungsi.

Yang sudah disertakan

Kelas Status
ReadableStream Lengkap
ReadableStreamDefaultController Lengkap
ReadableByteStreamController Lengkap
ReadableStreamDefaultReader Lengkap
ReadableStreamBYOBReader Lengkap
ReadableStreamBYOBRequest Lengkap
WritableStream Lengkap
WritableStreamDefaultController Lengkap
WritableStreamDefaultWriter Lengkap
TransformStream Lengkap
TransformStreamDefaultController Lengkap
ByteLengthQueuingStrategy Lengkap
CountQueuingStrategy Lengkap

Ditambah metode yang sering digunakan: pipeTo, pipeThrough, tee, getReader, iterasi async, Symbol.asyncIterator, pembacaan BYOB dengan { min }.

Menghasilkan data

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

Mengonsumsi data

Tiga cara membaca:

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

Mentransformasi

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

Aliran byte (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 } didukung — read(view, { min: 2 }) hanya selesai ketika setidaknya 2 elemen tersedia.

Tekanan balik

desiredSize dan ready berfungsi sesuai deskripsi spesifikasi. Gunakan strategi antrean untuk memberi produsen saluran umpan balik:

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

Dengan 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 untuk caching

Pola umum: baca body dua kali.

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

Response.clone() melakukan ini untuk Anda dan mengembalikan Response baru.

Catatan penting

  • Produsen sinkron saja. pull() yang melakukan I/O PHP pemblokiran akan memblokir skrip. Gunakan hook transport fetch untuk HTTP; untuk file, sebaiknya baca ke memori terlebih dahulu. setTimeout / setInterval tersedia (lihat Event loop), tetapi mesin menggunakan waktu virtual secara default — setTimeout(cb, 1000) di dalam pull() tidak tidur selama PHP, melainkan memajukan waktu loop. Untuk pengaturan tempo waktu nyata, host harus menjalankan tickEventLoop() berdasarkan jam nyata.

Lihat juga