Signatur
Beschreibung
Das Interface ReadableStreamDefaultController gehört zur Streams API. Eine Instanz lässt sich nicht selbst erzeugen: Der Konstruktor new ReadableStream(underlyingSource) legt den Controller automatisch an und übergibt ihn an die Callbacks start(controller) und pull(controller) der zugrunde liegenden Quelle.
Über den Controller steuert die Datenquelle den Stream:
controller.enqueue(chunk): reiht einen Chunk in die interne Warteschlange ein.controller.close(): schließt den Stream. Bereits eingereihte Chunks können danach noch gelesen werden.controller.error(e): versetzt den Stream in den Fehlerzustand. Alle ausstehenden und künftigen Lesevorgänge werden dann miteabgelehnt.controller.desiredSize(nur lesbar): gibt an, wie viel Platz noch in der Warteschlange ist, gemessen an derhighWaterMarkder Queuing-Strategie. Diese Angabe ist die Grundlage für Gegendruck (Backpressure).
Du brauchst den Controller immer dann, wenn du eigene Datenquellen als Stream bereitstellen willst, etwa Events, WebSocket-Nachrichten, erzeugte Daten oder zerlegte Dateien. Bei reinen Byte-Quellen mit Zero-Copy-Unterstützung (type: "bytes") bekommst du stattdessen einen ReadableByteStreamController.
Man unterscheidet zwei Arten von Quellen. Eine Push-Quelle ruft enqueue() selbstständig auf, sobald Daten da sind, und sollte dabei desiredSize beachten. Eine Pull-Quelle liefert Daten nur in pull(). Diese Methode ruft der Stream automatisch auf, sobald desiredSize > 0 ist.
Parameter
| Name | Typ | Default | Beschreibung |
|---|---|---|---|
| $chunk | any | Parameter von enqueue(chunk): beliebiger Wert, der in die Warteschlange eingereiht wird, z. B. ein String, ein Uint8Array oder ein Objekt. |
|
| $e | any | undefined | Parameter von error(e): der Fehlergrund, mit dem Leser abgewiesen werden. Am besten übergibst du ein Error-Objekt. |
Beispiele
Stream in start() mit festen Daten füllen und lesen
const stream = new ReadableStream({
start(controller) {
['Hallo', ' ', 'Welt'].forEach((teil) => controller.enqueue(teil));
controller.close(); // eingereihte Chunks bleiben lesbar
}
});
const leseAlles = async () => {
let text = '';
for await (const chunk of stream) text += chunk;
console.log(text);
};
leseAlles();Pull-Quelle mit Gegendruck: Zähler, der nur bei Bedarf liefert
let n = 0;
const zaehler = new ReadableStream({
pull(controller) {
// pull wird nur aufgerufen, wenn desiredSize > 0 ist
n += 1;
controller.enqueue(n);
if (n === 3) controller.close();
}
}, { highWaterMark: 1 });
const reader = zaehler.getReader();
const lies = async () => {
while (true) {
const { value, done } = await reader.read();
if (done) break;
console.log('Wert:', value);
}
};
lies();Push-Quelle aus Events mit desiredSize und error()
const ticker = new ReadableStream({
start(controller) {
let i = 0;
this.timer = setInterval(() => {
i += 1;
if (controller.desiredSize <= 0) {
console.log('Warteschlange voll, Tick verworfen');
return;
}
controller.enqueue(`Tick ${i}`);
if (i === 5) {
clearInterval(this.timer);
controller.error(new Error('Quelle getrennt'));
}
}, 100);
},
cancel(reason) {
clearInterval(this.timer); // Aufräumen bei reader.cancel()
console.log('Abgebrochen:', reason);
}
}, { highWaterMark: 10 });
const lesen = async () => {
try {
for await (const t of ticker) console.log(t);
} catch (err) {
console.error('Stream-Fehler:', err.message);
}
};
lesen();// Wichtig · Fallstricke
- Kein eigener Konstruktor:
new ReadableStreamDefaultController()wirft einenTypeError. Den Controller bekommst du nur über die Callbacks vonReadableStream. - Nach close() oder error() wirft ein weiterer Aufruf von
enqueue()oderclose()einenTypeError. Prüfe deshalb den eigenen Zustand oderdesiredSize !== null, bevor du nach dem Schließen weitere Daten einreihst. - error() verwirft die Warteschlange: Anders als bei
close()gehen Chunks, die noch nicht gelesen wurden, sofort verloren. In Beispiel 3 kommt deshalb Tick 5 nicht mehr an. - Gegendruck wird nicht erzwungen:
enqueue()funktioniert auch beidesiredSize <= 0. Beachtet die Quelle den Wert nicht, wächst der Speicherverbrauch unbegrenzt. - pull() und Promises: Gibt
pull()ein Promise zurück, ruft der Streampull()erst wieder auf, wenn dieses Promise erfüllt ist. Wird es abgelehnt, geht der Stream in den Fehlerzustand. - Kompatibilität: Der Controller wird in allen modernen Browsern unterstützt. Die asynchrone Iteration mit
for awaitüber einenReadableStreamist jünger; Safari unterstützt sie erst spät. In Node.js ist der Controller unterstream/webverfügbar, ab Version 18 auch global.