Signatur
Beschreibung
Mit enqueue() übergibt die Quelle eines ReadableStream Daten an den Stream. Der Controller wird der Quelle in den Callbacks start(controller) und pull(controller) des Underlying-Source-Objekts übergeben. Jeder Aufruf legt genau einen Chunk in die interne Warteschlange. Von dort holt ihn ein Leser (reader.read(), for await...of, pipeTo()) ab.
Ein Chunk kann ein beliebiger JavaScript-Wert sein: Strings, Objekte, Uint8Array usw. Wartet gerade ein Leser mit einem offenen read(), bekommt er den Chunk direkt. Sonst bleibt der Chunk in der Warteschlange, und controller.desiredSize sinkt um die Größe, die die queuingStrategy für ihn berechnet.
enqueue() wirft selbst keinen Druck zurück, der Aufruf gelingt also auch bei voller Warteschlange. Um Rückstau (Backpressure) zu beachten, sollte die Quelle desiredSize prüfen oder Daten nur in pull() liefern. pull() ruft der Stream erst auf, wenn wieder Platz ist.
Parameter
| Name | Typ | Default | Beschreibung |
|---|---|---|---|
| $chunk Pflicht | any | Der einzureihende Datenblock. Bei Byte-Streams (type: "bytes") wird stattdessen ReadableByteStreamController mit einer ArrayBufferView genutzt. |
Rückgabewert
undefined).Beispiele
Einfacher Stream aus festen Werten
const stream = new ReadableStream({
start(controller) {
controller.enqueue('Hallo');
controller.enqueue('Welt');
controller.close();
}
});
for await (const chunk of stream) {
console.log(chunk);
}Pull-basierter Zähler mit Backpressure
let i = 0;
const zaehler = new ReadableStream({
pull(controller) {
// pull() wird nur aufgerufen, wenn der Stream mehr Daten will
if (i < 3) {
controller.enqueue(i++);
} else {
controller.close();
}
}
}, { highWaterMark: 1 });
const reader = zaehler.getReader();
let ergebnis;
while (!(ergebnis = await reader.read()).done) {
console.log('Wert:', ergebnis.value);
}Timer-Ereignisse als Stream mit Abbruch
let timer;
const ticks = new ReadableStream({
start(controller) {
let n = 0;
timer = setInterval(() => {
controller.enqueue(`Tick ${++n}`);
if (n === 3) {
clearInterval(timer);
controller.close();
}
}, 100);
},
cancel() {
// wichtig: nach cancel() kein enqueue() mehr
clearInterval(timer);
}
});
const text = [];
for await (const t of ticks) text.push(t);
console.log(text.join(', '));Text als Uint8Array für eine Response einreihen
const encoder = new TextEncoder();
const body = new ReadableStream({
start(controller) {
controller.enqueue(encoder.encode('{"status":'));
controller.enqueue(encoder.encode('"ok"}'));
controller.close();
}
});
const daten = await new Response(body).json();
console.log(daten.status);// Wichtig · Fallstricke
- TypeError:
enqueue()wirft einenTypeError, wenn der Stream schon geschlossen wurde (close()), einen Fehlerzustand hat (error()) oder vom Konsumenten abgebrochen wurde (cancel()). Asynchrone Quellen wie Timer, WebSockets oder Event-Listener sollten deshalb imcancel()-Callback aufräumen. - Backpressure:
enqueue()nimmt immer an, auch wenndesiredSize <= 0ist. Wer das ignoriert, füllt bei schnellen Quellen den Speicher. Prüfecontroller.desiredSizeoder liefere Daten nur inpull(). - Response-Bodys: Für
new Response(stream)oderfetch-Uploads müssen die ChunksUint8Arraysein. Strings führen beim Lesen zu einemTypeError. - Byte-Streams: Bei
type: "bytes"bekommt die Quelle einenReadableByteStreamControllerstatt einesReadableStreamDefaultController. - Kompatibilität: Wird von allen modernen Browsern unterstützt.
for await...ofdirekt über einenReadableStreamfunktioniert in Safari erst seit neueren Versionen. Zur SicherheitgetReader()verwenden. In Node.js steht die API global ab Version 18 bereit, vorher überstream/web.