İçeriğe geç

Akışlar

Markdown olarak görüntüle GitHub üzerinde düzenle Dokümantasyon: 5b53f69

Phasis, WHATWG Streams Standard'ın tamamını uygular. Her sınıf, her okuyucu modu, her yardımcı. Node ve modern tarayıcılarda çalışan kodun aynısı Phasis'te çalışır.

Uygulama, fetch'ın dahili olarak kullandığı uygulamadır — Response.body gerçek bir ReadableStream, Request akış gövdelerini kabul eder ve Response'u klonlamak için gereken tee() çalışır.

Sunulanlar

Sınıf Durum
ReadableStream Tam
ReadableStreamDefaultController Tam
ReadableByteStreamController Tam
ReadableStreamDefaultReader Tam
ReadableStreamBYOBReader Tam
ReadableStreamBYOBRequest Tam
WritableStream Tam
WritableStreamDefaultController Tam
WritableStreamDefaultWriter Tam
TransformStream Tam
TransformStreamDefaultController Tam
ByteLengthQueuingStrategy Tam
CountQueuingStrategy Tam

Sık kullandığınız yöntemler de mevcut: pipeTo, pipeThrough, tee, getReader, asenkron yineleme, Symbol.asyncIterator, { min } ile BYOB okuma.

Veri üretme

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

Veri tüketme

Okumanın üç yolu:

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

Dönüştürme

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

Bayt akışları (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 } desteklenir — read(view, { min: 2 }) yalnızca en az 2 öğe olduğunda çözümlenir.

Geri basınç

desiredSize ve ready, spesifikasyonda açıklandığı gibi çalışır. Üreticiye geri bildirim kanalı sağlamak için bir kuyruk stratejisi kullanın:

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

fetch ile

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

Önbellekleme için tee

Yaygın örüntü: gövdeyi iki kez okumak.

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

Response.clone() bunu sizin için yapar ve yeni bir Response döndürür.

Dikkat edilmesi gerekenler

  • Yalnızca eşzamanlı üreticiler. Engelleyici PHP I/O işlemleri yapan bir pull(), betiği engeller. HTTP için fetch aktarım kancasını kullanın; dosyalarda ise önceden belleğe okumayı tercih edin. setTimeout / setInterval kullanılabilir (bkz. Olay döngüsü), ancak motor varsayılan olarak sanal zaman kullanır: bir pull() içindeki setTimeout(cb, 1000), PHP sürecini uyutmaz; döngü saatini ilerletir. Gerçek zamanlı ilerleme için ana uygulamanın tickEventLoop() çağrılarını gerçek saate göre yürütmesi gerekir.

Ayrıca bkz.