Node.js v26.0.0 Dokumentation

Stream#

Ein Stream ist eine abstrakte Schnittstelle für die Arbeit mit Streaming-Daten in Node.js. Das node:stream-Modul stellt eine API zur Implementierung der Stream-Schnittstelle bereit.

Es gibt viele von Node.js bereitgestellte Stream-Objekte. Zum Beispiel sind eine Anfrage an einen HTTP-Server und process.stdout beides Stream-Instanzen.

Streams können lesbar, schreibbar oder beides sein. Alle Streams sind Instanzen des EventEmitter.

Zugriff auf das node:stream-Modul

const stream = require('node:stream');
copy

Das node:stream-Modul ist nützlich für die Erstellung neuer Arten von Stream-Instanzen. Es ist normalerweise nicht notwendig, das node:stream-Modul zu verwenden, um Streams zu konsumieren.

Gliederung dieses Dokuments#

Dieses Dokument enthält zwei Hauptabschnitte und einen dritten Abschnitt für Hinweise. Der erste Abschnitt erläutert, wie bestehende Streams innerhalb einer Anwendung verwendet werden. Der zweite Abschnitt erklärt, wie man neue Arten von Streams erstellt.

Arten von Streams#

Es gibt vier grundlegende Stream-Typen innerhalb von Node.js

• Writable: Streams, in die Daten geschrieben werden können (z. B. fs.createWriteStream()).
• Readable: Streams, aus denen Daten gelesen werden können (z. B. fs.createReadStream()).
• Duplex: Streams, die sowohl Readable als auch Writable sind (z. B. net.Socket).
• Transform: Duplex-Streams, die die Daten beim Schreiben und Lesen modifizieren oder transformieren können (z. B. zlib.createDeflate()).

Zusätzlich enthält dieses Modul die Hilfsfunktionen stream.duplexPair(), stream.pipeline(), stream.finished(), stream.Readable.from() und stream.addAbortSignal().

Streams Promises-API#

Die stream/promises-API bietet einen alternativen Satz asynchroner Hilfsfunktionen für Streams, die Promise-Objekte zurückgeben, anstatt Callbacks zu verwenden. Die API ist über require('node:stream/promises') oder require('node:stream').promises zugänglich.

stream.pipeline(streams[, options])#

stream.pipeline(source[, ...transforms], destination[, options])#

• streams <Stream[]> | <Iterable[]> | <AsyncIterable[]> | <Function[]>
• source <Stream> | <Iterable> | <AsyncIterable> | <Function>
  • Rückgabe: <Promise> | <AsyncIterable>
• ...transforms <Stream> | <Function>
  • source <AsyncIterable>
  • Rückgabe: <Promise> | <AsyncIterable>
• destination <Stream> | <Function>
  • source <AsyncIterable>
  • Rückgabe: <Promise> | <AsyncIterable>
• options <Object> Pipeline-Optionen
  • signal <AbortSignal>
  • end <boolean> Beende den Ziel-Stream, wenn der Quell-Stream endet. Transform-Streams werden immer beendet, selbst wenn dieser Wert false ist. Standard: true.
• Rückgabe: <Promise> Wird erfüllt, wenn die Pipeline abgeschlossen ist.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

await pipeline(
  createReadStream('archive.tar'),
  createGzip(),
  createWriteStream('archive.tar.gz'),
);
console.log('Pipeline succeeded.');
copy

Um ein AbortSignal zu verwenden, übergeben Sie es innerhalb eines Optionen-Objekts als letztes Argument. Wenn das Signal abgebrochen wird, wird destroy für die zugrunde liegende Pipeline mit einem AbortError aufgerufen.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  const ac = new AbortController();
  const signal = ac.signal;

  setImmediate(() => ac.abort());
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
    { signal },
  );
}

run().catch(console.error); // AbortError
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';
import { createGzip } from 'node:zlib';

const ac = new AbortController();
const { signal } = ac;
setImmediate(() => ac.abort());
try {
  await pipeline(
    createReadStream('archive.tar'),
    createGzip(),
    createWriteStream('archive.tar.gz'),
    { signal },
  );
} catch (err) {
  console.error(err); // AbortError
}
copy

Die pipeline-API unterstützt auch asynchrone Generatoren

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    fs.createReadStream('lowercase.txt'),
    async function* (source, { signal }) {
      source.setEncoding('utf8');  // Work with strings rather than `Buffer`s.
      for await (const chunk of source) {
        yield await processChunk(chunk, { signal });
      }
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';

await pipeline(
  createReadStream('lowercase.txt'),
  async function* (source, { signal }) {
    source.setEncoding('utf8');  // Work with strings rather than `Buffer`s.
    for await (const chunk of source) {
      yield await processChunk(chunk, { signal });
    }
  },
  createWriteStream('uppercase.txt'),
);
console.log('Pipeline succeeded.');
copy

Denken Sie daran, das an den asynchronen Generator übergebene signal-Argument zu verarbeiten. Besonders in dem Fall, in dem der asynchrone Generator die Quelle für die Pipeline (d. h. das erste Argument) ist oder die Pipeline niemals abgeschlossen wird.

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    async function* ({ signal }) {
      await someLongRunningfn({ signal });
      yield 'asd';
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);
import { pipeline } from 'node:stream/promises';
import fs from 'node:fs';
await pipeline(
  async function* ({ signal }) {
    await someLongRunningfn({ signal });
    yield 'asd';
  },
  fs.createWriteStream('uppercase.txt'),
);
console.log('Pipeline succeeded.');
copy

Die pipeline-API bietet eine Callback-Version

stream.finished(stream[, options])#

• stream <Stream> | <ReadableStream> | <WritableStream> Ein lesbarer und/oder schreibbarer Stream/Webstream.
• options <Object>
  • error <boolean> | <undefined>
  • readable <boolean> | <undefined>
  • writable <boolean> | <undefined>
  • signal <AbortSignal> | <undefined>
  • cleanup <boolean> | <undefined> Wenn true, werden die von dieser Funktion registrierten Listener entfernt, bevor das Promise erfüllt wird. Standard: false.
• Rückgabe: <Promise> Wird erfüllt, wenn der Stream nicht mehr lesbar oder schreibbar ist.

const { finished } = require('node:stream/promises');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // Drain the stream.
import { finished } from 'node:stream/promises';
import { createReadStream } from 'node:fs';

const rs = createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // Drain the stream.
copy

Die finished-API bietet auch eine Callback-Version.

stream.finished() hinterlässt hängende Ereignis-Listener (insbesondere 'error', 'end', 'finish' und 'close'), nachdem das zurückgegebene Promise aufgelöst oder abgelehnt wurde. Der Grund dafür ist, dass unerwartete 'error'-Ereignisse (aufgrund fehlerhafter Stream-Implementierungen) keine unerwarteten Abstürze verursachen sollen. Wenn dies unerwünscht ist, sollte options.cleanup auf true gesetzt werden.

await finished(rs, { cleanup: true });
copy

Objektmodus#

Alle von Node.js-APIs erstellten Streams operieren ausschließlich auf Strings, <Buffer>, <TypedArray> und <DataView> Objekten.

• Strings und Buffers sind die am häufigsten mit Streams verwendeten Typen.
• TypedArray und DataView ermöglichen die Verarbeitung binärer Daten mit Typen wie Int32Array oder Uint8Array. Wenn Sie ein TypedArray oder DataView in einen Stream schreiben, verarbeitet Node.js die Rohbytes.

Es ist jedoch möglich, dass Stream-Implementierungen mit anderen Arten von JavaScript-Werten arbeiten (mit Ausnahme von null, das innerhalb von Streams eine besondere Bedeutung hat). Solche Streams gelten als im "Objektmodus" arbeitend.

Stream-Instanzen werden bei ihrer Erstellung mithilfe der objectMode-Option in den Objektmodus geschaltet. Der Versuch, einen bestehenden Stream in den Objektmodus zu schalten, ist nicht sicher.

Pufferung#

Sowohl Writable- als auch Readable-Streams speichern Daten in einem internen Puffer.

Die Menge der potenziell gepufferten Daten hängt von der highWaterMark-Option ab, die an den Konstruktor des Streams übergeben wird. Bei normalen Streams gibt die highWaterMark-Option eine Gesamtanzahl von Bytes an. Bei Streams, die im Objektmodus arbeiten, gibt die highWaterMark eine Gesamtanzahl von Objekten an. Bei Streams, die auf Strings operieren (aber diese nicht dekodieren), gibt die highWaterMark eine Gesamtanzahl von UTF-16-Code-Einheiten an.

Daten werden in Readable-Streams gepuffert, wenn die Implementierung stream.push(chunk) aufruft. Wenn der Consumer des Streams stream.read() nicht aufruft, verbleiben die Daten in der internen Warteschlange, bis sie konsumiert werden.

Sobald die Gesamtgröße des internen Lesepuffers den durch highWaterMark angegebenen Schwellenwert erreicht, stoppt der Stream vorübergehend das Lesen von Daten aus der zugrunde liegenden Ressource, bis die aktuell gepufferten Daten konsumiert werden können (das heißt, der Stream hört auf, die interne readable._read()-Methode aufzurufen, die zum Füllen des Lesepuffers verwendet wird).

Daten werden in Writable-Streams gepuffert, wenn die Methode writable.write(chunk) wiederholt aufgerufen wird. Solange die Gesamtgröße des internen Schreibpuffers unter dem durch highWaterMark gesetzten Schwellenwert liegt, geben Aufrufe von writable.write() den Wert true zurück. Sobald die Größe des internen Puffers den highWaterMark erreicht oder überschreitet, wird false zurückgegeben.

Ein Hauptziel der stream-API, insbesondere der Methode stream.pipe(), ist es, die Pufferung von Daten auf ein akzeptables Maß zu begrenzen, sodass Quellen und Ziele unterschiedlicher Geschwindigkeit den verfügbaren Speicher nicht überfordern.

Die highWaterMark-Option ist ein Schwellenwert, kein Limit: Sie diktiert die Menge an Daten, die ein Stream puffert, bevor er aufhört, nach weiteren Daten zu fragen. Sie erzwingt im Allgemeinen keine strikte Speicherbeschränkung. Spezifische Stream-Implementierungen können sich dafür entscheiden, strengere Grenzwerte durchzusetzen, aber dies ist optional.

Da Duplex- und Transform-Streams sowohl Readable als auch Writable sind, unterhält jeder zwei separate interne Puffer zum Lesen und Schreiben. Dies ermöglicht es jeder Seite, unabhängig von der anderen zu arbeiten und gleichzeitig einen angemessenen und effizienten Datenfluss aufrechtzuerhalten. Zum Beispiel sind net.Socket-Instanzen Duplex-Streams, deren Readable-Seite den Verbrauch von vom Socket empfangenen Daten ermöglicht und deren Writable-Seite das Schreiben von Daten auf den Socket erlaubt. Da Daten möglicherweise schneller oder langsamer auf den Socket geschrieben als empfangen werden können, sollte jede Seite unabhängig von der anderen arbeiten (und puffern).

Die Mechanik der internen Pufferung ist ein internes Implementierungsdetail und kann jederzeit geändert werden. Für bestimmte fortgeschrittene Implementierungen können die internen Puffer jedoch mithilfe von writable.writableBuffer oder readable.readableBuffer abgerufen werden. Von der Verwendung dieser nicht dokumentierten Eigenschaften wird abgeraten.

API für Stream-Consumer#

Fast alle Node.js-Anwendungen, egal wie einfach sie sind, verwenden auf irgendeine Weise Streams. Das Folgende ist ein Beispiel für die Verwendung von Streams in einer Node.js-Anwendung, die einen HTTP-Server implementiert.

const http = require('node:http');

const server = http.createServer((req, res) => {
  // `req` is an http.IncomingMessage, which is a readable stream.
  // `res` is an http.ServerResponse, which is a writable stream.

  let body = '';
  // Get the data as utf8 strings.
  // If an encoding is not set, Buffer objects will be received.
  req.setEncoding('utf8');

  // Readable streams emit 'data' events once a listener is added.
  req.on('data', (chunk) => {
    body += chunk;
  });

  // The 'end' event indicates that the entire body has been received.
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // Write back something interesting to the user:
      res.write(typeof data);
      res.end();
    } catch (er) {
      // uh oh! bad json!
      res.statusCode = 400;
      return res.end(`error: ${er.message}`);
    }
  });
});

server.listen(1337);

// $ curl localhost:1337 -d "{}"
// object
// $ curl localhost:1337 -d "\"foo\""
// string
// $ curl localhost:1337 -d "not json"
// error: Unexpected token 'o', "not json" is not valid JSON
copy

Writable-Streams (wie res im Beispiel) stellen Methoden wie write() und end() bereit, die verwendet werden, um Daten in den Stream zu schreiben.

Readable-Streams verwenden die EventEmitter-API, um Anwendungscode zu benachrichtigen, wenn Daten zum Lesen aus dem Stream verfügbar sind. Diese verfügbaren Daten können auf verschiedene Weise aus dem Stream gelesen werden.

Sowohl Writable- als auch Readable-Streams verwenden die EventEmitter-API auf verschiedene Weise, um den aktuellen Status des Streams zu kommunizieren.

Duplex- und Transform-Streams sind sowohl Writable als auch Readable.

Anwendungen, die Daten in einen Stream schreiben oder aus diesem konsumieren, sind nicht verpflichtet, die Stream-Schnittstellen direkt zu implementieren, und haben im Allgemeinen keinen Grund, require('node:stream') aufzurufen.

Entwickler, die neue Arten von Streams implementieren möchten, sollten sich auf den Abschnitt API für Stream-Implementierer beziehen.

Schreibbare Streams#

Schreibbare Streams sind eine Abstraktion für ein Ziel, in das Daten geschrieben werden.

Beispiele für Writable-Streams sind:

• HTTP-Anfragen auf der Client-Seite
• HTTP-Antworten auf der Server-Seite
• fs-Schreib-Streams
• zlib-Streams
• crypto-Streams
• TCP-Sockets
• Standardeingabe (stdin) eines Kindprozesses
• process.stdout, process.stderr

Einige dieser Beispiele sind eigentlich Duplex-Streams, die die Writable-Schnittstelle implementieren.

Alle Writable-Streams implementieren die durch die Klasse stream.Writable definierte Schnittstelle.

Während sich spezifische Instanzen von Writable-Streams in verschiedener Weise unterscheiden können, folgen alle Writable-Streams demselben grundlegenden Nutzungsmuster, wie im folgenden Beispiel illustriert:

const myStream = getWritableStreamSomehow();
myStream.write('some data');
myStream.write('some more data');
myStream.end('done writing data');
copy

Klasse: stream.Writable#

Ereignis: 'close'#

Das 'close'-Ereignis wird emittiert, wenn der Stream und alle seine zugrunde liegenden Ressourcen (z. B. ein Dateideskriptor) geschlossen wurden. Das Ereignis zeigt an, dass keine weiteren Ereignisse mehr emittiert werden und keine weiteren Berechnungen stattfinden werden.

Ein Writable-Stream emittiert immer das 'close'-Ereignis, wenn er mit der emitClose-Option erstellt wurde.

Ereignis: 'drain'#

Wenn ein Aufruf von stream.write(chunk) den Wert false zurückgibt, wird das 'drain'-Ereignis emittiert, wenn es angebracht ist, das Schreiben von Daten in den Stream fortzusetzen.

// Write the data to the supplied writable stream one million times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  write();
  function write() {
    let ok = true;
    do {
      i--;
      if (i === 0) {
        // Last time!
        writer.write(data, encoding, callback);
      } else {
        // See if we should continue, or wait.
        // Don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // Had to stop early!
      // Write some more once it drains.
      writer.once('drain', write);
    }
  }
}
copy

Ereignis: 'error'#

• Typ: <Error>

Das 'error'-Ereignis wird emittiert, wenn beim Schreiben oder Pipelining von Daten ein Fehler aufgetreten ist. Der Listener-Callback erhält bei Aufruf ein einzelnes Error-Argument.

Der Stream wird geschlossen, wenn das 'error'-Ereignis emittiert wird, es sei denn, die Option autoDestroy wurde beim Erstellen des Streams auf false gesetzt.

Nach 'error' sollten keine weiteren Ereignisse außer 'close' emittiert werden (einschließlich 'error'-Ereignisse).

Ereignis: 'finish'#

Das 'finish'-Ereignis wird emittiert, nachdem die Methode stream.end() aufgerufen wurde und alle Daten an das zugrunde liegende System geleert (flushed) wurden.

const writer = getWritableStreamSomehow();
for (let i = 0; i < 100; i++) {
  writer.write(`hello, #${i}!\n`);
}
writer.on('finish', () => {
  console.log('All writes are now complete.');
});
writer.end('This is the end\n');
copy

Ereignis: 'pipe'#

• src <stream.Readable> Quell-Stream, der in diesen Writable-Stream gepiped wird.

Das 'pipe'-Ereignis wird emittiert, wenn die Methode stream.pipe() auf einem lesbaren Stream aufgerufen wird und diesen Writable-Stream zu seinen Zielen hinzufügt.

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('pipe', (src) => {
  console.log('Something is piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer);
copy

Ereignis: 'unpipe'#

• src <stream.Readable> Der Quell-Stream, der diesen Writable-Stream unpiped (die Verbindung trennte).

Das 'unpipe'-Ereignis wird emittiert, wenn die Methode stream.unpipe() auf einem Readable-Stream aufgerufen wird und diesen Writable-Stream aus seinen Zielen entfernt.

Dies wird auch emittiert, falls dieser Writable-Stream einen Fehler emittiert, während ein Readable-Stream in ihn hineinpiped.

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('unpipe', (src) => {
  console.log('Something has stopped piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer);
copy

writable.cork()#

Die Methode writable.cork() erzwingt, dass alle geschriebenen Daten im Speicher gepuffert werden. Die gepufferten Daten werden geleert, wenn entweder die Methode stream.uncork() oder stream.end() aufgerufen wird.

Die primäre Absicht von writable.cork() ist es, Situationen zu bewältigen, in denen mehrere kleine Datenstücke in schneller Folge in den Stream geschrieben werden. Anstatt sie sofort an das zugrunde liegende Ziel weiterzuleiten, puffert writable.cork() alle Datenstücke, bis writable.uncork() aufgerufen wird, was diese dann alle an writable._writev() übergibt, falls vorhanden. Dies verhindert eine Head-of-Line-Blocking-Situation, in der Daten gepuffert werden, während auf die Verarbeitung des ersten kleinen Datenstücks gewartet wird. Die Verwendung von writable.cork() ohne die Implementierung von writable._writev() kann jedoch nachteilige Auswirkungen auf den Durchsatz haben.

Siehe auch: writable.uncork(), writable._writev().

writable.destroy([error])#

• error <Error> Optional, ein Fehler, der mit dem 'error'-Ereignis emittiert werden soll.
• Rückgabe: <this>

Zerstöre den Stream. Emittiere optional ein 'error'-Ereignis und ein 'close'-Ereignis (außer emitClose ist auf false gesetzt). Nach diesem Aufruf ist der schreibbare Stream beendet, und nachfolgende Aufrufe von write() oder end() führen zu einem ERR_STREAM_DESTROYED-Fehler. Dies ist eine destruktive und sofortige Methode, um einen Stream zu zerstören. Frühere Aufrufe von write() wurden möglicherweise noch nicht geleert und können einen ERR_STREAM_DESTROYED-Fehler auslösen. Verwenden Sie end() anstelle von destroy(), wenn Daten vor dem Schließen geleert werden sollen, oder warten Sie auf das 'drain'-Ereignis, bevor Sie den Stream zerstören.

const { Writable } = require('node:stream');

const myStream = new Writable();

const fooErr = new Error('foo error');
myStream.destroy(fooErr);
myStream.on('error', (fooErr) => console.error(fooErr.message)); // foo error
copy

const { Writable } = require('node:stream');

const myStream = new Writable();

myStream.destroy();
myStream.on('error', function wontHappen() {});
copy

const { Writable } = require('node:stream');

const myStream = new Writable();
myStream.destroy();

myStream.write('foo', (error) => console.error(error.code));
// ERR_STREAM_DESTROYED
copy

Sobald destroy() aufgerufen wurde, sind alle weiteren Aufrufe No-Ops, und außer von _destroy() sollten keine weiteren Fehler mehr als 'error' emittiert werden.

Implementierer sollten diese Methode nicht überschreiben, sondern stattdessen writable._destroy() implementieren.

writable.closed#

• Typ: <boolean>

Ist true, nachdem 'close' emittiert wurde.

writable.destroyed#

• Typ: <boolean>

Ist true, nachdem writable.destroy() aufgerufen wurde.

const { Writable } = require('node:stream');

const myStream = new Writable();

console.log(myStream.destroyed); // false
myStream.destroy();
console.log(myStream.destroyed); // true
copy

writable.end([chunk[, encoding]][, callback])#

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> Optionale Daten zum Schreiben. Für Streams, die nicht im Objektmodus operieren, muss chunk ein <string>, <Buffer>, <TypedArray> oder <DataView> sein. Für Streams im Objektmodus kann chunk jeder beliebige JavaScript-Wert außer null sein.
• encoding <string> Die Kodierung, wenn chunk ein String ist.
• callback <Function> Callback für den Abschluss des Streams.
• Rückgabe: <this>

Der Aufruf der Methode writable.end() signalisiert, dass keine weiteren Daten mehr in den Writable geschrieben werden. Die optionalen chunk- und encoding-Argumente erlauben es, ein letztes zusätzliches Datenstück unmittelbar vor dem Schließen des Streams zu schreiben.

Der Aufruf der Methode stream.write() nach dem Aufruf von stream.end() löst einen Fehler aus.

// Write 'hello, ' and then end with 'world!'.
const fs = require('node:fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed!
copy

writable.setDefaultEncoding(encoding)#

• encoding <string> Die neue Standardkodierung.
• Rückgabe: <this>

Die Methode writable.setDefaultEncoding() setzt die Standard-encoding für einen Writable-Stream.

writable.uncork()#

Die Methode writable.uncork() leert alle seit dem Aufruf von stream.cork() gepufferten Daten.

Bei der Verwendung von writable.cork() und writable.uncork() zur Verwaltung der Pufferung von Schreibvorgängen in einen Stream sollten Aufrufe von writable.uncork() mit process.nextTick() verzögert werden. Dies ermöglicht die Stapelverarbeitung (Batching) aller writable.write()-Aufrufe, die innerhalb einer bestimmten Node.js-Ereignisschleifenphase auftreten.

stream.cork();
stream.write('some ');
stream.write('data ');
process.nextTick(() => stream.uncork());
copy

Wenn die Methode writable.cork() mehrfach auf einem Stream aufgerufen wird, muss die gleiche Anzahl an Aufrufen von writable.uncork() erfolgen, um die gepufferten Daten zu leeren.

stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
  stream.uncork();
  // The data will not be flushed until uncork() is called a second time.
  stream.uncork();
});
copy

Siehe auch: writable.cork().

writable.writable#

• Typ: <boolean>

Ist true, wenn es sicher ist, writable.write() aufzurufen, was bedeutet, dass der Stream weder zerstört noch fehlerhaft oder beendet wurde.

writable.writableAborted#

• Typ: <boolean>

Gibt zurück, ob der Stream zerstört wurde oder einen Fehler hatte, bevor 'finish' emittiert wurde.

writable.writableEnded#

• Typ: <boolean>

Ist true, nachdem writable.end() aufgerufen wurde. Diese Eigenschaft gibt nicht an, ob die Daten bereits geleert wurden; verwenden Sie dafür stattdessen writable.writableFinished.

writable.writableCorked#

• Typ: <integer>

Anzahl der Male, die writable.uncork() aufgerufen werden muss, um den Stream vollständig zu entkorken.

writable.errored#

• Typ: <Error>

Gibt einen Fehler zurück, wenn der Stream mit einem Fehler zerstört wurde.

writable.writableFinished#

• Typ: <boolean>

Wird unmittelbar vor dem Emittieren des 'finish'-Ereignisses auf true gesetzt.

writable.writableHighWaterMark#

• Typ: <number>

Gibt den Wert von highWaterMark zurück, der bei der Erstellung dieses Writable-Streams übergeben wurde.

writable.writableLength#

• Typ: <number>

Diese Eigenschaft enthält die Anzahl der Bytes (oder Objekte) in der Warteschlange, die bereit sind, geschrieben zu werden. Der Wert liefert Introspektionsdaten zum Status des highWaterMark.

writable.writableNeedDrain#

• Typ: <boolean>

Ist true, wenn der Puffer des Streams voll war und der Stream ein 'drain'-Ereignis emittieren wird.

writable.writableObjectMode#

• Typ: <boolean>

Getter für die Eigenschaft objectMode eines gegebenen Writable-Streams.

writable[Symbol.asyncDispose]()#

Ruft writable.destroy() mit einem AbortError auf und gibt ein Promise zurück, das erfüllt wird, wenn der Stream beendet ist.

writable.write(chunk[, encoding][, callback])#

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> Optionale Daten zum Schreiben. Für Streams, die nicht im Objektmodus operieren, muss chunk ein <string>, <Buffer>, <TypedArray> oder <DataView> sein. Für Streams im Objektmodus kann chunk jeder beliebige JavaScript-Wert außer null sein.
• encoding <string> | <null> Die Kodierung, falls chunk ein String ist. Standard: 'utf8'
• callback <Function> Callback für den Zeitpunkt, an dem dieses Datenstück geleert wurde.
• Rückgabe: <boolean> false, wenn der Stream möchte, dass der aufrufende Code auf das Eintreten des Ereignisses 'drain' wartet, bevor er weitere Daten schreibt; andernfalls true.

Die Methode writable.write() schreibt Daten in den Stream und ruft den angegebenen callback auf, sobald die Daten vollständig verarbeitet wurden. Wenn ein Fehler auftritt, wird der callback mit dem Fehler als erstem Argument aufgerufen. Der callback wird asynchron und vor dem Emittieren von 'error' aufgerufen.

Der Rückgabewert ist true, wenn der interne Puffer nach der Aufnahme von chunk kleiner als der bei der Erstellung des Streams konfigurierte highWaterMark ist. Wenn false zurückgegeben wird, sollten weitere Versuche, Daten in den Stream zu schreiben, gestoppt werden, bis das 'drain'-Ereignis emittiert wird.

Während ein Stream nicht "draint", puffern Aufrufe von write() den chunk und geben false zurück. Sobald alle aktuell gepufferten Stücke geleert (vom Betriebssystem zur Auslieferung akzeptiert) wurden, wird das 'drain'-Ereignis emittiert. Sobald write() false zurückgibt, schreiben Sie keine weiteren Stücke, bis das 'drain'-Ereignis emittiert wird. Während das Aufrufen von write() auf einem Stream, der nicht draint, erlaubt ist, puffert Node.js alle geschriebenen Stücke, bis der maximale Speicherverbrauch erreicht ist, woraufhin es bedingungslos abbricht. Selbst vor dem Abbruch führt ein hoher Speicherverbrauch zu einer schlechten Garbage-Collector-Leistung und einem hohen RSS (das typischerweise nicht an das System zurückgegeben wird, selbst wenn der Speicher nicht mehr benötigt wird). Da TCP-Sockets möglicherweise niemals drainen, wenn der entfernte Partner die Daten nicht liest, kann das Schreiben in einen Socket, der nicht draint, zu einer remote ausnutzbaren Sicherheitslücke führen.

Das Schreiben von Daten, während der Stream nicht draint, ist besonders problematisch für einen Transform-Stream, da diese standardmäßig pausiert sind, bis sie gepiped werden oder ein 'data'- oder 'readable'-Ereignis-Listener hinzugefügt wird.

Wenn die zu schreibenden Daten bei Bedarf generiert oder abgerufen werden können, wird empfohlen, die Logik in einen Readable-Stream zu kapseln und stream.pipe() zu verwenden. Wenn das Aufrufen von write() jedoch bevorzugt wird, ist es möglich, Backpressure zu respektieren und Speicherprobleme durch Verwendung des 'drain'-Ereignisses zu vermeiden.

function write(data, cb) {
  if (!stream.write(data)) {
    stream.once('drain', cb);
  } else {
    process.nextTick(cb);
  }
}

// Wait for cb to be called before doing any other write.
write('hello', () => {
  console.log('Write completed, do more writes now.');
});
copy

Ein Writable-Stream im Objektmodus ignoriert das encoding-Argument immer.

Lesbare Streams#

Lesbare Streams sind eine Abstraktion für eine Quelle, aus der Daten konsumiert werden.

Beispiele für Readable-Streams sind:

• HTTP-Antworten auf der Client-Seite
• HTTP-Anfragen auf der Server-Seite
• fs-Lese-Streams
• zlib-Streams
• crypto-Streams
• TCP-Sockets
• stdout und stderr von Kindprozessen
• process.stdin

Alle Readable-Streams implementieren die Schnittstelle der Klasse stream.Readable.

Zwei Lesemodi#

Readable-Streams arbeiten effektiv in einem von zwei Modi: fließend (flowing) und pausiert (paused). Diese Modi sind vom Objektmodus getrennt. Ein Readable-Stream kann sich im Objektmodus befinden oder nicht, unabhängig davon, ob er sich im fließenden oder pausierten Modus befindet.

• Im fließenden Modus werden Daten automatisch aus dem zugrunde liegenden System gelesen und einer Anwendung so schnell wie möglich über Ereignisse mittels der EventEmitter-Schnittstelle bereitgestellt.

• Im pausierten Modus muss die Methode stream.read() explizit aufgerufen werden, um Datenstücke aus dem Stream zu lesen.

Alle Readable-Streams beginnen im pausierten Modus, können aber auf eine der folgenden Weisen in den fließenden Modus geschaltet werden:

• Hinzufügen eines 'data'-Ereignis-Listeners.
• Aufrufen der Methode stream.resume().
• Aufrufen der Methode stream.pipe(), um die Daten an einen Writable-Stream zu senden.

Der Readable-Stream kann auf eine der folgenden Weisen zurück in den pausierten Modus schalten:

• Wenn keine Pipe-Ziele existieren, durch Aufrufen der Methode stream.pause().
• Wenn Pipe-Ziele existieren, durch Entfernen aller Pipe-Ziele. Mehrere Pipe-Ziele können durch Aufrufen der Methode stream.unpipe() entfernt werden.

Das wichtige Konzept, das man sich merken muss, ist, dass ein Readable-Stream keine Daten generiert, bis ein Mechanismus zum Konsumieren oder Ignorieren dieser Daten bereitgestellt wird. Wenn der konsumierende Mechanismus deaktiviert oder entfernt wird, versucht der Readable-Stream, die Generierung der Daten zu stoppen.

Aus Gründen der Abwärtskompatibilität führt das Entfernen von 'data'-Ereignis-Listenern nicht automatisch zum Pausieren des Streams. Wenn zudem Pipe-Ziele vorhanden sind, garantiert das Aufrufen von stream.pause() nicht, dass der Stream pausiert bleibt, sobald diese Ziele drainen und nach weiteren Daten fragen.

Wenn ein Readable in den fließenden Modus geschaltet wird und keine Consumer zur Verarbeitung der Daten verfügbar sind, gehen diese Daten verloren. Dies kann beispielsweise passieren, wenn die Methode readable.resume() aufgerufen wird, ohne dass ein Listener an das 'data'-Ereignis angehängt ist, oder wenn ein 'data'-Ereignis-Listener vom Stream entfernt wird.

Das Hinzufügen eines 'readable'-Ereignis-Listeners bewirkt automatisch, dass der Stream aufhört zu fließen, und die Daten müssen dann über readable.read() konsumiert werden. Wenn der 'readable'-Ereignis-Listener entfernt wird, beginnt der Stream wieder zu fließen, sofern ein 'data'-Ereignis-Listener vorhanden ist.

Drei Zustände#

Die "zwei Modi" des Betriebs für einen Readable-Stream sind eine vereinfachte Abstraktion für die kompliziertere interne Zustandsverwaltung, die innerhalb der Readable-Stream-Implementierung stattfindet.

Genauer gesagt befindet sich jeder Readable-Stream zu jedem Zeitpunkt in einem von drei möglichen Zuständen:

• readable.readableFlowing === null
• readable.readableFlowing === false
• readable.readableFlowing === true

Wenn readable.readableFlowing den Wert null hat, ist kein Mechanismus zum Konsumieren der Daten des Streams bereitgestellt. Daher generiert der Stream keine Daten. Während er sich in diesem Zustand befindet, schaltet das Anhängen eines Listeners für das 'data'-Ereignis, das Aufrufen der readable.pipe()-Methode oder das Aufrufen der readable.resume()-Methode readable.readableFlowing auf true, wodurch der Readable-Stream beginnt, aktiv Ereignisse zu emittieren, während Daten generiert werden.

Das Aufrufen von readable.pause(), readable.unpipe() oder das Empfangen von Backpressure führt dazu, dass readable.readableFlowing auf false gesetzt wird, wodurch der Fluss von Ereignissen vorübergehend angehalten, aber nicht die Generierung von Daten gestoppt wird. Während er sich in diesem Zustand befindet, schaltet das Anhängen eines Listeners für das 'data'-Ereignis readable.readableFlowing nicht auf true.

const { PassThrough, Writable } = require('node:stream');
const pass = new PassThrough();
const writable = new Writable();

pass.pipe(writable);
pass.unpipe(writable);
// readableFlowing is now false.

pass.on('data', (chunk) => { console.log(chunk.toString()); });
// readableFlowing is still false.
pass.write('ok');  // Will not emit 'data'.
pass.resume();     // Must be called to make stream emit 'data'.
// readableFlowing is now true.
copy

Während readable.readableFlowing den Wert false hat, können sich Daten im internen Puffer des Streams ansammeln.

Wählen Sie einen API-Stil#

Die Readable-Stream-API hat sich über mehrere Node.js-Versionen hinweg weiterentwickelt und bietet mehrere Methoden zum Konsumieren von Stream-Daten. Im Allgemeinen sollten Entwickler eine Methode zum Konsumieren von Daten wählen und niemals mehrere Methoden verwenden, um Daten aus einem einzigen Stream zu konsumieren. Insbesondere die Verwendung einer Kombination aus on('data'), on('readable'), pipe() oder asynchronen Iteratoren könnte zu unintuitivem Verhalten führen.

Klasse: stream.Readable#

Ereignis: 'close'#

Das 'close'-Ereignis wird emittiert, wenn der Stream und alle seine zugrunde liegenden Ressourcen (z. B. ein Dateideskriptor) geschlossen wurden. Das Ereignis zeigt an, dass keine weiteren Ereignisse mehr emittiert werden und keine weiteren Berechnungen stattfinden werden.

Ein Readable-Stream emittiert immer das 'close'-Ereignis, wenn er mit der emitClose-Option erstellt wurde.

Ereignis: 'data'#

• chunk <Buffer> | <string> | <any> Das Datenstück. Für Streams, die nicht im Objektmodus operieren, ist das Stück entweder ein String oder ein Buffer. Für Streams im Objektmodus kann das Stück jeder JavaScript-Wert außer null sein.

Das 'data'-Ereignis wird jedes Mal emittiert, wenn der Stream die Kontrolle über ein Datenstück an einen Consumer abgibt. Dies kann geschehen, wenn der Stream in den fließenden Modus geschaltet wird, indem readable.pipe() oder readable.resume() aufgerufen wird oder ein Listener-Callback an das 'data'-Ereignis angehängt wird. Das 'data'-Ereignis wird auch jedes Mal emittiert, wenn die Methode readable.read() aufgerufen wird und ein Datenstück zur Rückgabe verfügbar ist.

Das Anhängen eines 'data'-Ereignis-Listeners an einen Stream, der nicht explizit pausiert wurde, schaltet den Stream in den fließenden Modus. Daten werden dann weitergegeben, sobald sie verfügbar sind.

Der Listener-Callback erhält das Datenstück als String, wenn für den Stream eine Standardkodierung mithilfe der readable.setEncoding()-Methode festgelegt wurde; andernfalls werden die Daten als Buffer übergeben.

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
});
copy

Ereignis: 'end'#

Das 'end'-Ereignis wird emittiert, wenn keine Daten mehr aus dem Stream konsumiert werden können.

Das 'end'-Ereignis wird nicht emittiert, sofern die Daten nicht vollständig konsumiert wurden. Dies kann erreicht werden, indem der Stream in den fließenden Modus geschaltet wird oder stream.read() wiederholt aufgerufen wird, bis alle Daten konsumiert wurden.

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
});
readable.on('end', () => {
  console.log('There will be no more data.');
});
copy

Ereignis: 'error'#

• Typ: <Error>

Das 'error'-Ereignis kann von einer Readable-Implementierung jederzeit emittiert werden. Typischerweise kann dies auftreten, wenn der zugrunde liegende Stream aufgrund eines internen Fehlers keine Daten generieren kann oder wenn eine Stream-Implementierung versucht, ein ungültiges Datenstück zu pushen.

Dem Listener-Callback wird ein einzelnes Error-Objekt übergeben.

Ereignis: 'pause'#

Das 'pause'-Ereignis wird emittiert, wenn stream.pause() aufgerufen wird und readableFlowing nicht false ist.

Ereignis: 'readable'#

Das 'readable'-Ereignis wird emittiert, wenn Daten aus dem Stream gelesen werden können, bis hin zum konfigurierten High Water Mark (state.highWaterMark). Effektiv zeigt es an, dass der Stream neue Informationen im Puffer hat. Wenn in diesem Puffer Daten verfügbar sind, kann stream.read() aufgerufen werden, um diese Daten abzurufen. Zusätzlich kann das 'readable'-Ereignis auch emittiert werden, wenn das Ende des Streams erreicht wurde.

const readable = getReadableStreamSomehow();
readable.on('readable', function() {
  // There is some data to read now.
  let data;

  while ((data = this.read()) !== null) {
    console.log(data);
  }
});
copy

Wenn das Ende des Streams erreicht wurde, gibt der Aufruf von stream.read() den Wert null zurück und löst das 'end'-Ereignis aus. Dies gilt auch, wenn nie Daten zum Lesen vorhanden waren. Im folgenden Beispiel ist foo.txt zum Beispiel eine leere Datei:

const fs = require('node:fs');
const rr = fs.createReadStream('foo.txt');
rr.on('readable', () => {
  console.log(`readable: ${rr.read()}`);
});
rr.on('end', () => {
  console.log('end');
});
copy

Die Ausgabe beim Ausführen dieses Skripts ist:

$ node test.js
readable: null
end
copy

In einigen Fällen führt das Anhängen eines Listeners für das 'readable'-Ereignis dazu, dass eine gewisse Menge an Daten in einen internen Puffer gelesen wird.

Im Allgemeinen sind die readable.pipe()- und 'data'-Ereignis-Mechanismen einfacher zu verstehen als das 'readable'-Ereignis. Die Handhabung von 'readable' könnte jedoch zu einem erhöhten Durchsatz führen.

Wenn sowohl 'readable' als auch 'data' gleichzeitig verwendet werden, hat 'readable' Vorrang bei der Steuerung des Flusses, d. h. 'data' wird nur emittiert, wenn stream.read() aufgerufen wird. Die Eigenschaft readableFlowing wird dann zu false. Wenn 'data'-Listener vorhanden sind, während 'readable' entfernt wird, beginnt der Stream zu fließen, d. h. 'data'-Ereignisse werden ohne Aufruf von .resume() emittiert.

Ereignis: 'resume'#

Das 'resume'-Ereignis wird emittiert, wenn stream.resume() aufgerufen wird und readableFlowing nicht true ist.

readable.destroy([error])#

• error <Error> Fehler, der als Payload im 'error'-Ereignis übergeben wird.
• Rückgabe: <this>

Zerstöre den Stream. Emittiere optional ein 'error'-Ereignis und ein 'close'-Ereignis (außer emitClose ist auf false gesetzt). Nach diesem Aufruf gibt der lesbare Stream alle internen Ressourcen frei und nachfolgende Aufrufe von push() werden ignoriert.

Sobald destroy() aufgerufen wurde, sind alle weiteren Aufrufe No-Ops, und außer von _destroy() sollten keine weiteren Fehler mehr als 'error' emittiert werden.

Implementierer sollten diese Methode nicht überschreiben, sondern stattdessen readable._destroy() implementieren.

readable.closed#

• Typ: <boolean>

Ist true, nachdem 'close' emittiert wurde.

readable.destroyed#

• Typ: <boolean>

Ist true, nachdem readable.destroy() aufgerufen wurde.

readable.isPaused()#

• Rückgabewert: <boolean>

Die Methode readable.isPaused() gibt den aktuellen Betriebszustand des Readable-Streams zurück. Dies wird hauptsächlich von dem Mechanismus verwendet, der der readable.pipe()-Methode zugrunde liegt. In den meisten typischen Fällen gibt es keinen Grund, diese Methode direkt zu verwenden.

const readable = new stream.Readable();

readable.isPaused(); // === false
readable.pause();
readable.isPaused(); // === true
readable.resume();
readable.isPaused(); // === false
copy

readable.pause()#

• Rückgabe: <this>

Die Methode readable.pause() bewirkt, dass ein Stream im fließenden Modus aufhört, 'data'-Ereignisse zu emittieren, und aus dem fließenden Modus schaltet. Alle Daten, die verfügbar werden, verbleiben im internen Puffer.

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  readable.pause();
  console.log('There will be no additional data for 1 second.');
  setTimeout(() => {
    console.log('Now data will start flowing again.');
    readable.resume();
  }, 1000);
});
copy

Die Methode readable.pause() hat keine Wirkung, wenn ein 'readable'-Ereignis-Listener vorhanden ist.

readable.pipe(destination[, options])#

• destination <stream.Writable> Das Ziel für das Schreiben von Daten.
• options <Object> Pipe-Optionen.
  • end <boolean> Beende den Writer, wenn der Reader endet. Standard: true.
• Rückgabe: <stream.Writable> Das Ziel, was eine Kette von Pipes ermöglicht, wenn es ein Duplex- oder Transform-Stream ist.

Die Methode readable.pipe() hängt einen Writable-Stream an den Readable-Stream an, wodurch dieser automatisch in den fließenden Modus schaltet und alle seine Daten an den angehängten Writable-Stream weiterleitet. Der Datenfluss wird automatisch verwaltet, sodass der Ziel-Writable-Stream nicht von einem schnelleren Readable-Stream überfordert wird.

Das folgende Beispiel piped alle Daten vom readable in eine Datei namens file.txt:

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt'.
readable.pipe(writable);
copy

Es ist möglich, mehrere Writable-Streams an einen einzigen Readable-Stream anzuhängen.

Die Methode readable.pipe() gibt eine Referenz auf den Ziel-Stream zurück, wodurch es möglich ist, Ketten von gepipten Streams einzurichten.

const fs = require('node:fs');
const zlib = require('node:zlib');
const r = fs.createReadStream('file.txt');
const z = zlib.createGzip();
const w = fs.createWriteStream('file.txt.gz');
r.pipe(z).pipe(w);
copy

Standardmäßig wird stream.end() auf dem Ziel-Writable-Stream aufgerufen, wenn der Quell-Readable-Stream 'end' emittiert, sodass das Ziel nicht mehr beschreibbar ist. Um dieses Standardverhalten zu deaktivieren, kann die Option end auf false gesetzt werden, wodurch der Ziel-Stream offen bleibt.

reader.pipe(writer, { end: false });
reader.on('end', () => {
  writer.end('Goodbye\n');
});
copy

Ein wichtiger Hinweis ist, dass der Writable-Ziel-Stream nicht automatisch geschlossen wird, wenn der Readable-Stream während der Verarbeitung einen Fehler emittiert. Wenn ein Fehler auftritt, ist es notwendig, jeden Stream manuell zu schließen, um Speicherlecks zu vermeiden.

Die process.stderr- und process.stdout Writable-Streams werden niemals geschlossen, bevor der Node.js-Prozess beendet wird, unabhängig von den angegebenen Optionen.

readable.read([size])#

• size <number> Optionales Argument zur Angabe der zu lesenden Datenmenge.
• Rückgabe: <string> | <Buffer> | <null> | <any>

Die Methode readable.read() liest Daten aus dem internen Puffer und gibt sie zurück. Wenn keine Daten zum Lesen verfügbar sind, wird null zurückgegeben. Standardmäßig werden die Daten als Buffer-Objekt zurückgegeben, sofern keine Kodierung mithilfe der readable.setEncoding()-Methode angegeben wurde oder der Stream im Objektmodus arbeitet.

Das optionale size-Argument gibt eine spezifische Anzahl von Bytes zum Lesen an. Wenn size Bytes nicht zum Lesen verfügbar sind, wird null zurückgegeben, außer der Stream wurde bereits beendet, in welchem Fall alle im internen Puffer verbliebenen Daten zurückgegeben werden.

Wenn das size-Argument nicht angegeben ist, werden alle im internen Puffer enthaltenen Daten zurückgegeben.

Das size-Argument muss kleiner oder gleich 1 GiB sein.

Die Methode readable.read() sollte nur auf Readable-Streams aufgerufen werden, die im pausierten Modus arbeiten. Im fließenden Modus wird readable.read() automatisch aufgerufen, bis der interne Puffer vollständig geleert ist.

const readable = getReadableStreamSomehow();

// 'readable' may be triggered multiple times as data is buffered in
readable.on('readable', () => {
  let chunk;
  console.log('Stream is readable (new data received in buffer)');
  // Use a loop to make sure we read all currently available data
  while (null !== (chunk = readable.read())) {
    console.log(`Read ${chunk.length} bytes of data...`);
  }
});

// 'end' will be triggered once when there is no more data available
readable.on('end', () => {
  console.log('Reached end of stream.');
});
copy

Jeder Aufruf von readable.read() gibt ein Datenstück oder null zurück, was signalisiert, dass in diesem Moment keine weiteren Daten zum Lesen vorhanden sind. Diese Stücke werden nicht automatisch verkettet. Da ein einzelner read()-Aufruf nicht alle Daten zurückgibt, kann eine while-Schleife erforderlich sein, um kontinuierlich Datenstücke zu lesen, bis alle Daten abgerufen wurden. Beim Lesen einer großen Datei könnte .read() vorübergehend null zurückgeben, was darauf hinweist, dass der gesamte gepufferte Inhalt konsumiert wurde, aber möglicherweise noch weitere Daten zu puffern sind. In solchen Fällen wird ein neues 'readable'-Ereignis emittiert, sobald weitere Daten im Puffer sind, und das 'end'-Ereignis signalisiert das Ende der Datenübertragung.

Um also den gesamten Inhalt einer Datei aus einem readable-Stream zu lesen, ist es notwendig, Datenstücke über mehrere 'readable'-Ereignisse hinweg zu sammeln:

const chunks = [];

readable.on('readable', () => {
  let chunk;
  while (null !== (chunk = readable.read())) {
    chunks.push(chunk);
  }
});

readable.on('end', () => {
  const content = chunks.join('');
});
copy

Ein Readable-Stream im Objektmodus gibt bei einem Aufruf von readable.read(size) immer ein einzelnes Element zurück, unabhängig vom Wert des size-Arguments.

Wenn die Methode readable.read() ein Datenstück zurückgibt, wird auch ein 'data'-Ereignis emittiert.

Das Aufrufen von stream.read([size]) nach dem Emittieren des 'end'-Ereignisses gibt null zurück. Es wird kein Laufzeitfehler ausgelöst.

readable.readable#

• Typ: <boolean>

Ist true, wenn es sicher ist, readable.read() aufzurufen, was bedeutet, dass der Stream nicht zerstört wurde und weder 'error' noch 'end' emittiert hat.

readable.readableAborted#

• Typ: <boolean>

Gibt zurück, ob der Stream zerstört wurde oder einen Fehler hatte, bevor 'end' emittiert wurde.

readable.readableDidRead#

• Typ: <boolean>

Gibt zurück, ob 'data' emittiert wurde.

readable.readableEncoding#

• Typ: <null> | <string>

Getter für die Eigenschaft encoding eines gegebenen Readable-Streams. Die encoding-Eigenschaft kann mithilfe der readable.setEncoding()-Methode gesetzt werden.

readable.readableEnded#

• Typ: <boolean>

Wird true, wenn das 'end'-Ereignis emittiert wird.

readable.errored#

• Typ: <Error>

Gibt einen Fehler zurück, wenn der Stream mit einem Fehler zerstört wurde.

readable.readableFlowing#

• Typ: <boolean>

Diese Eigenschaft spiegelt den aktuellen Zustand eines Readable-Streams wider, wie im Abschnitt Drei Zustände beschrieben.

readable.readableHighWaterMark#

• Typ: <number>

Gibt den Wert von highWaterMark zurück, der bei der Erstellung dieses Readable-Streams übergeben wurde.

readable.readableLength#

• Typ: <number>

Diese Eigenschaft enthält die Anzahl der Bytes (oder Objekte) in der Warteschlange, die bereit zum Lesen sind. Der Wert liefert Introspektionsdaten zum Status des highWaterMark.

readable.readableObjectMode#

• Typ: <boolean>

Getter für die Eigenschaft objectMode eines gegebenen Readable-Streams.

readable.resume()#

• Rückgabe: <this>

Die Methode readable.resume() bewirkt, dass ein explizit pausierter Readable-Stream wieder 'data'-Ereignisse emittiert und in den fließenden Modus schaltet.

Die Methode readable.resume() kann verwendet werden, um die Daten eines Streams vollständig zu konsumieren, ohne diese Daten tatsächlich zu verarbeiten.

getReadableStreamSomehow()
  .resume()
  .on('end', () => {
    console.log('Reached the end, but did not read anything.');
  });
copy

Die Methode readable.resume() hat keine Wirkung, wenn ein 'readable'-Ereignis-Listener vorhanden ist.

readable.setEncoding(encoding)#

• encoding <string> Die zu verwendende Kodierung.
• Rückgabe: <this>

Die Methode readable.setEncoding() legt die Zeichenkodierung für Daten fest, die aus dem Readable-Stream gelesen werden.

Standardmäßig ist keine Kodierung zugewiesen, und Stream-Daten werden als Buffer-Objekte zurückgegeben. Das Festlegen einer Kodierung bewirkt, dass Stream-Daten als Strings der angegebenen Kodierung anstelle von Buffer-Objekten zurückgegeben werden. Zum Beispiel führt der Aufruf von readable.setEncoding('utf8') dazu, dass die Ausgabedaten als UTF-8 interpretiert und als Strings übergeben werden. Der Aufruf von readable.setEncoding('hex') bewirkt, dass die Daten im hexadezimalen String-Format kodiert werden.

Der Readable-Stream handhabt korrekt Multibyte-Zeichen, die durch den Stream übertragen werden, welche andernfalls falsch dekodiert würden, wenn sie einfach als Buffer-Objekte aus dem Stream abgerufen würden.

const readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', (chunk) => {
  assert.equal(typeof chunk, 'string');
  console.log('Got %d characters of string data:', chunk.length);
});
copy

readable.unpipe([destination])#

• destination <stream.Writable> Optionaler spezifischer Stream, dessen Pipe getrennt werden soll.
• Rückgabe: <this>

Die Methode readable.unpipe() trennt einen Writable-Stream, der zuvor mithilfe der stream.pipe()-Methode angehängt wurde.

Wenn das destination-Argument nicht angegeben ist, werden alle Pipes getrennt.

Wenn das destination-Argument angegeben ist, aber keine Pipe für diesen Stream eingerichtet ist, bewirkt die Methode nichts.

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second.
readable.pipe(writable);
setTimeout(() => {
  console.log('Stop writing to file.txt.');
  readable.unpipe(writable);
  console.log('Manually close the file stream.');
  writable.end();
}, 1000);
copy

readable.unshift(chunk[, encoding])#

• chunk <Buffer> | <TypedArray> | <DataView> | <string> | <null> | <any> Datenstück, das auf die Lesewarteschlange zurückgelegt (unshift) werden soll. Für Streams, die nicht im Objektmodus arbeiten, muss chunk ein <string>, <Buffer>, <TypedArray>, <DataView> oder null sein. Für Streams im Objektmodus kann chunk jeder JavaScript-Wert sein.
• encoding <string> Kodierung von String-Datenstücken. Muss eine gültige Buffer-Kodierung sein, wie z. B. 'utf8' oder 'ascii'.

Das Übergeben von null als chunk signalisiert das Ende des Streams (EOF) und verhält sich identisch mit readable.push(null), wonach keine weiteren Daten mehr geschrieben werden können. Das EOF-Signal wird an das Ende des Puffers gestellt, und alle gepufferten Daten werden dennoch geleert.

Die Methode readable.unshift() schiebt ein Datenstück zurück in den internen Puffer. Dies ist nützlich in Situationen, in denen ein Stream von Code konsumiert wird, der einige der bereits optimistisch aus der Quelle gelesenen Daten "ent-konsumieren" muss, damit diese an eine andere Partei weitergegeben werden können.

Die Methode stream.unshift(chunk) kann nicht aufgerufen werden, nachdem das 'end'-Ereignis emittiert wurde; andernfalls wird ein Laufzeitfehler geworfen.

Entwickler, die stream.unshift() verwenden, sollten oft in Erwägung ziehen, stattdessen einen Transform-Stream zu verwenden. Siehe den Abschnitt API für Stream-Implementierer für weitere Informationen.

// Pull off a header delimited by \n\n.
// Use unshift() if we get too much.
// Call the callback with (error, header, stream).
const { StringDecoder } = require('node:string_decoder');
function parseHeader(stream, callback) {
  stream.on('error', callback);
  stream.on('readable', onReadable);
  const decoder = new StringDecoder('utf8');
  let header = '';
  function onReadable() {
    let chunk;
    while (null !== (chunk = stream.read())) {
      const str = decoder.write(chunk);
      if (str.includes('\n\n')) {
        // Found the header boundary.
        const split = str.split(/\n\n/);
        header += split.shift();
        const remaining = split.join('\n\n');
        const buf = Buffer.from(remaining, 'utf8');
        stream.removeListener('error', callback);
        // Remove the 'readable' listener before unshifting.
        stream.removeListener('readable', onReadable);
        if (buf.length)
          stream.unshift(buf);
        // Now the body of the message can be read from the stream.
        callback(null, header, stream);
        return;
      }
      // Still reading the header.
      header += str;
    }
  }
}
copy

Anders als stream.push(chunk) beendet stream.unshift(chunk) den Lesevorgang nicht durch das Zurücksetzen des internen Lesestatus des Streams. Dies kann unerwartete Ergebnisse verursachen, wenn readable.unshift() während eines Lesevorgangs aufgerufen wird (d. h. innerhalb einer stream._read()-Implementierung eines benutzerdefinierten Streams). Das Folgen des Aufrufs von readable.unshift() mit einem unmittelbaren stream.push('') setzt den Lesestatus korrekt zurück, es ist jedoch am besten, den Aufruf von readable.unshift() während eines laufenden Lesevorgangs einfach zu vermeiden.

readable.wrap(stream)#

• stream <Stream> Ein "old-style" lesbarer Stream.
• Rückgabe: <this>

Vor Node.js 0.10 implementierten Streams nicht die gesamte node:stream-Modul-API, wie sie derzeit definiert ist. (Siehe Kompatibilität für weitere Informationen.)

Bei der Verwendung einer älteren Node.js-Bibliothek, die 'data'-Ereignisse emittiert und eine stream.pause()-Methode hat, die nur beratend ist, kann die Methode readable.wrap() verwendet werden, um einen Readable-Stream zu erstellen, der den alten Stream als Datenquelle verwendet.

Es wird selten notwendig sein, readable.wrap() zu verwenden, aber die Methode wurde als Komfortfunktion für die Interaktion mit älteren Node.js-Anwendungen und -Bibliotheken bereitgestellt.

const { OldReader } = require('./old-api-module.js');
const { Readable } = require('node:stream');
const oreader = new OldReader();
const myReader = new Readable().wrap(oreader);

myReader.on('readable', () => {
  myReader.read(); // etc.
});
copy

readable[Symbol.asyncIterator]()#

• Rückgabe: <AsyncIterator>, um den Stream vollständig zu konsumieren.

const fs = require('node:fs');

async function print(readable) {
  readable.setEncoding('utf8');
  let data = '';
  for await (const chunk of readable) {
    data += chunk;
  }
  console.log(data);
}

print(fs.createReadStream('file')).catch(console.error);
copy

Wenn die Schleife mit break, return oder einem throw endet, wird der Stream zerstört. Mit anderen Worten: Das Iterieren über einen Stream konsumiert den Stream vollständig. Der Stream wird in Stücken gelesen, die der Größe der highWaterMark-Option entsprechen. Im Codebeispiel oben sind die Daten in einem einzigen Stück, wenn die Datei weniger als 64 KiB Daten enthält, da an fs.createReadStream() keine highWaterMark-Option übergeben wurde.

readable[Symbol.asyncDispose]()#

Ruft readable.destroy() mit einem AbortError auf und gibt ein Promise zurück, das erfüllt wird, wenn der Stream beendet ist.

readable.compose(stream[, options])#

• stream <Writable> | <Duplex> | <WritableStream> | <TransformStream> | <Function>
• options <Object>
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Duplex> ein mit dem Stream stream zusammengesetzter Stream.

import { Readable } from 'node:stream';

async function* splitToWords(source) {
  for await (const chunk of source) {
    const words = String(chunk).split(' ');

    for (const word of words) {
      yield word;
    }
  }
}

const wordsStream = Readable.from(['text passed through', 'composed stream']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // prints ['text', 'passed', 'through', 'composed', 'stream']
copy

readable.compose(s) ist äquivalent zu stream.compose(readable, s).

Diese Methode ermöglicht es auch, ein <AbortSignal> bereitzustellen, das den zusammengesetzten Stream bei Abbruch zerstört.

Weitere Informationen finden Sie unter stream.compose(...streams).

readable.iterator([options])#

• options <Object>
  • destroyOnReturn <boolean> Wenn auf false gesetzt, wird der Stream nicht zerstört, wenn return für den Async-Iterator aufgerufen wird oder eine for await...of-Iteration mittels break, return oder throw verlassen wird. Standard: true.
• Rückgabe: <AsyncIterator> zum Konsumieren des Streams.

Der von dieser Methode erstellte Iterator gibt Benutzern die Möglichkeit, die Zerstörung des Streams zu stornieren, wenn die for await...of-Schleife durch return, break oder throw verlassen wird, oder wenn der Iterator den Stream zerstören soll, falls der Stream während der Iteration einen Fehler ausgegeben hat.

const { Readable } = require('node:stream');

async function printIterator(readable) {
  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // false

  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // Will print 2 and then 3
  }

  console.log(readable.destroyed); // True, stream was totally consumed
}

async function printSymbolAsyncIterator(readable) {
  for await (const chunk of readable) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // true
}

async function showBoth() {
  await printIterator(Readable.from([1, 2, 3]));
  await printSymbolAsyncIterator(Readable.from([1, 2, 3]));
}

showBoth();
copy

readable.map(fn[, options])#

• fn <Function> | <AsyncFunction> eine Funktion, die auf jedes Datenpaket (Chunk) im Stream angewendet wird.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• options <Object>
  • concurrency <number> die maximale Anzahl gleichzeitiger Aufrufe von fn für den Stream. Standard: 1.
  • highWaterMark <number> wie viele Elemente zwischengespeichert werden sollen, während auf die Konsumierung der zugeordneten Elemente gewartet wird. Standard: concurrency * 2 - 1.
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Readable> ein mit der Funktion fn zugeordneter Stream.

Diese Methode ermöglicht das Abbilden (Mapping) über den Stream. Die Funktion fn wird für jedes Datenpaket im Stream aufgerufen. Wenn die Funktion fn ein Promise zurückgibt, wird dieses Promise awaited, bevor es an den Ergebnis-Stream weitergeleitet wird.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
  console.log(chunk); // 2, 4, 6, 8
}
// With an asynchronous mapper, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map((domain) => resolver.resolve4(domain), { concurrency: 2 });
for await (const result of dnsResults) {
  console.log(result); // Logs the DNS result of resolver.resolve4.
}
copy

readable.filter(fn[, options])#

• fn <Function> | <AsyncFunction> eine Funktion zum Filtern von Datenpaketen aus dem Stream.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• options <Object>
  • concurrency <number> die maximale Anzahl gleichzeitiger Aufrufe von fn für den Stream. Standard: 1.
  • highWaterMark <number> wie viele Elemente zwischengespeichert werden sollen, während auf die Konsumierung der gefilterten Elemente gewartet wird. Standard: concurrency * 2 - 1.
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Readable> ein mit dem Prädikat fn gefilterter Stream.

Diese Methode ermöglicht das Filtern des Streams. Für jedes Datenpaket im Stream wird die Funktion fn aufgerufen; wenn sie einen wahrheitsgemäßen (truthy) Wert zurückgibt, wird das Paket an den Ergebnis-Stream weitergegeben. Wenn die Funktion fn ein Promise zurückgibt, wird dieses Promise awaited.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).filter(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address.ttl > 60;
}, { concurrency: 2 });
for await (const result of dnsResults) {
  // Logs domains with more than 60 seconds on the resolved dns record.
  console.log(result);
}
copy

readable.forEach(fn[, options])#

• fn <Function> | <AsyncFunction> eine Funktion, die für jedes Datenpaket des Streams aufgerufen wird.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• options <Object>
  • concurrency <number> die maximale Anzahl gleichzeitiger Aufrufe von fn für den Stream. Standard: 1.
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Promise> ein Promise für den Zeitpunkt, an dem der Stream beendet ist.

Diese Methode ermöglicht das Iterieren über einen Stream. Für jedes Datenpaket im Stream wird die Funktion fn aufgerufen. Wenn die Funktion fn ein Promise zurückgibt, wird dieses Promise awaited.

Diese Methode unterscheidet sich von for await...of-Schleifen dadurch, dass sie Datenpakete optional gleichzeitig verarbeiten kann. Darüber hinaus kann eine forEach-Iteration nur gestoppt werden, indem eine signal-Option übergeben und der zugehörige AbortController abgebrochen wird, während for await...of mit break oder return gestoppt werden kann. In beiden Fällen wird der Stream zerstört.

Diese Methode unterscheidet sich vom Lauschen auf das 'data'-Ereignis dadurch, dass sie das readable-Ereignis in der zugrunde liegenden Mechanik verwendet und die Anzahl gleichzeitiger fn-Aufrufe begrenzen kann.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 });
await dnsResults.forEach((result) => {
  // Logs result, similar to `for await (const result of dnsResults)`
  console.log(result);
});
console.log('done'); // Stream has finished
copy

readable.toArray([options])#

• options <Object>
  • signal <AbortSignal> ermöglicht den Abbruch der toArray-Operation, wenn das Signal abgebrochen wird.
• Rückgabe: <Promise> ein Promise, das ein Array mit dem Inhalt des Streams enthält.

Diese Methode ermöglicht es, den Inhalt eines Streams einfach abzurufen.

Da diese Methode den gesamten Stream in den Speicher liest, macht sie die Vorteile von Streams zunichte. Sie ist für Interoperabilität und Komfort gedacht, nicht als primäre Methode zum Konsumieren von Streams.

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

await Readable.from([1, 2, 3, 4]).toArray(); // [1, 2, 3, 4]

const resolver = new Resolver();

// Make dns queries concurrently using .map and collect
// the results into an array using toArray
const dnsResults = await Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 }).toArray();
copy

readable.some(fn[, options])#

• fn <Function> | <AsyncFunction> eine Funktion, die für jedes Datenpaket des Streams aufgerufen wird.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• options <Object>
  • concurrency <number> die maximale Anzahl gleichzeitiger Aufrufe von fn für den Stream. Standard: 1.
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Promise> ein Promise, das zu true ausgewertet wird, wenn fn für mindestens eines der Datenpakete einen wahrheitsgemäßen Wert zurückgegeben hat.

Diese Methode ähnelt Array.prototype.some und ruft fn für jedes Datenpaket im Stream auf, bis der erwartete Rückgabewert true (oder ein wahrheitsgemäßer Wert) ist. Sobald der erwartete Rückgabewert eines fn-Aufrufs für ein Datenpaket wahrheitsgemäß ist, wird der Stream zerstört und das Promise mit true erfüllt. Wenn keiner der fn-Aufrufe für die Datenpakete einen wahrheitsgemäßen Wert zurückgibt, wird das Promise mit false erfüllt.

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).some((x) => x > 2); // true
await Readable.from([1, 2, 3, 4]).some((x) => x < 0); // false

// With an asynchronous predicate, making at most 2 file checks at a time.
const anyBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).some(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(anyBigFile); // `true` if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished
copy

readable.find(fn[, options])#

• fn <Function> | <AsyncFunction> eine Funktion, die für jedes Datenpaket des Streams aufgerufen wird.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• options <Object>
  • concurrency <number> die maximale Anzahl gleichzeitiger Aufrufe von fn für den Stream. Standard: 1.
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Promise> ein Promise, das zu dem ersten Datenpaket ausgewertet wird, für das fn einen wahrheitsgemäßen Wert zurückgegeben hat, oder undefined, wenn kein Element gefunden wurde.

Diese Methode ähnelt Array.prototype.find und ruft fn für jedes Datenpaket im Stream auf, um ein Paket mit einem wahrheitsgemäßen fn-Wert zu finden. Sobald der erwartete Rückgabewert eines fn-Aufrufs wahrheitsgemäß ist, wird der Stream zerstört und das Promise mit dem Wert erfüllt, für den fn einen wahrheitsgemäßen Wert zurückgegeben hat. Wenn alle fn-Aufrufe für die Datenpakete einen falschen Wert zurückgeben, wird das Promise mit undefined erfüllt.

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).find((x) => x > 2); // 3
await Readable.from([1, 2, 3, 4]).find((x) => x > 0); // 1
await Readable.from([1, 2, 3, 4]).find((x) => x > 10); // undefined

// With an asynchronous predicate, making at most 2 file checks at a time.
const foundBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).find(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(foundBigFile); // File name of large file, if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished
copy

readable.every(fn[, options])#

• fn <Function> | <AsyncFunction> eine Funktion, die für jedes Datenpaket des Streams aufgerufen wird.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• options <Object>
  • concurrency <number> die maximale Anzahl gleichzeitiger Aufrufe von fn für den Stream. Standard: 1.
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Promise> ein Promise, das zu true ausgewertet wird, wenn fn für alle Datenpakete einen wahrheitsgemäßen Wert zurückgegeben hat.

Diese Methode ähnelt Array.prototype.every und ruft fn für jedes Datenpaket im Stream auf, um zu prüfen, ob alle erwarteten Rückgabewerte für fn wahrheitsgemäß sind. Sobald der erwartete Rückgabewert eines fn-Aufrufs falsch ist, wird der Stream zerstört und das Promise mit false erfüllt. Wenn alle fn-Aufrufe für die Datenpakete einen wahrheitsgemäßen Wert zurückgeben, wird das Promise mit true erfüllt.

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).every((x) => x > 2); // false
await Readable.from([1, 2, 3, 4]).every((x) => x > 0); // true

// With an asynchronous predicate, making at most 2 file checks at a time.
const allBigFiles = await Readable.from([
  'file1',
  'file2',
  'file3',
]).every(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
// `true` if all files in the list are bigger than 1MiB
console.log(allBigFiles);
console.log('done'); // Stream has finished
copy

readable.flatMap(fn[, options])#

• fn <Function> | <AsyncGeneratorFunction> | <AsyncFunction> eine Funktion, die über jedes Datenpaket im Stream gemappt wird.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• options <Object>
  • concurrency <number> die maximale Anzahl gleichzeitiger Aufrufe von fn für den Stream. Standard: 1.
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Readable> ein mit der Funktion fn flach-abgebildeter Stream.

Diese Methode gibt einen neuen Stream zurück, indem der angegebene Callback auf jedes Datenpaket des Streams angewendet und das Ergebnis anschließend abgeflacht wird.

Es ist möglich, von fn einen Stream oder ein anderes iterierbares oder asynchron iterierbares Objekt zurückzugeben; die Ergebnis-Streams werden dann in den zurückgegebenen Stream zusammengeführt (abgeflacht).

import { Readable } from 'node:stream';
import { createReadStream } from 'node:fs';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
  console.log(chunk); // 1, 1, 2, 2, 3, 3, 4, 4
}
// With an asynchronous mapper, combine the contents of 4 files
const concatResult = Readable.from([
  './1.mjs',
  './2.mjs',
  './3.mjs',
  './4.mjs',
]).flatMap((fileName) => createReadStream(fileName));
for await (const result of concatResult) {
  // This will contain the contents (all chunks) of all 4 files
  console.log(result);
}
copy

readable.drop(limit[, options])#

• limit <number> die Anzahl der aus dem Readable zu löschenden Datenpakete.
• options <Object>
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Readable> ein Stream, bei dem limit Datenpakete gelöscht wurden.

Diese Methode gibt einen neuen Stream zurück, bei dem die ersten limit Datenpakete gelöscht wurden.

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).drop(2).toArray(); // [3, 4]
copy

readable.take(limit[, options])#

• limit <number> die Anzahl der aus dem Readable zu entnehmenden Datenpakete.
• options <Object>
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Readable> ein Stream, bei dem limit Datenpakete entnommen wurden.

Diese Methode gibt einen neuen Stream mit den ersten limit Datenpaketen zurück.

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).take(2).toArray(); // [1, 2]
copy

readable.reduce(fn[, initial[, options]])#

• fn <Function> | <AsyncFunction> eine Reduzierfunktion, die für jedes Datenpaket im Stream aufgerufen wird.
  • previous <any> der Wert, der aus dem letzten Aufruf von fn oder dem initial-Wert (falls angegeben) oder sonst dem ersten Datenpaket des Streams erhalten wurde.
  • data <any> ein Datenpaket aus dem Stream.
  • options <Object>
    • signal <AbortSignal> wird abgebrochen, wenn der Stream zerstört wird, wodurch der Aufruf von fn vorzeitig abgebrochen werden kann.
• initial <any> der anfängliche Wert für die Reduktion.
• options <Object>
  • signal <AbortSignal> ermöglicht die Zerstörung des Streams, wenn das Signal abgebrochen wird.
• Rückgabe: <Promise> ein Promise für den Endwert der Reduktion.

Diese Methode ruft fn nacheinander für jedes Datenpaket des Streams auf und übergibt das Ergebnis der Berechnung des vorherigen Elements. Sie gibt ein Promise für den Endwert der Reduktion zurück.

Wenn kein initial-Wert angegeben wird, wird das erste Datenpaket des Streams als Anfangswert verwendet. Wenn der Stream leer ist, wird das Promise mit einem TypeError mit dem Fehlercode ERR_INVALID_ARGS abgelehnt.

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .reduce(async (totalSize, file) => {
    const { size } = await stat(join(directoryPath, file));
    return totalSize + size;
  }, 0);

console.log(folderSize);
copy

Die Reduzierfunktion iteriert den Stream Element für Element, was bedeutet, dass es keinen concurrency-Parameter oder Parallelismus gibt. Um eine reduce-Operation gleichzeitig durchzuführen, können Sie die asynchrone Funktion in die Methode readable.map auslagern.

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .map((file) => stat(join(directoryPath, file)), { concurrency: 2 })
  .reduce((totalSize, { size }) => totalSize + size, 0);

console.log(folderSize);
copy

Duplex- und Transform-Streams#

Klasse: stream.Duplex#

Duplex-Streams sind Streams, die sowohl die Schnittstellen Readable als auch Writable implementieren.

Beispiele für Duplex-Streams sind

• TCP-Sockets
• zlib-Streams
• crypto-Streams

duplex.allowHalfOpen#

• Typ: <boolean>

Wenn false, beendet der Stream die Writable-Seite automatisch, wenn die Readable-Seite endet. Anfangs durch die allowHalfOpen-Konstruktoroption gesetzt, die standardmäßig true ist.

Dies kann manuell geändert werden, um das Half-Open-Verhalten einer bestehenden Duplex-Stream-Instanz zu ändern, muss jedoch vor dem Emittieren des 'end'-Ereignisses geschehen.

Klasse: stream.Transform#

Transform-Streams sind Duplex-Streams, bei denen die Ausgabe in irgendeiner Weise mit der Eingabe zusammenhängt. Wie alle Duplex-Streams implementieren Transform-Streams sowohl die Schnittstellen Readable als auch Writable.

Beispiele für Transform-Streams sind

• zlib-Streams
• crypto-Streams

transform.destroy([error])#

• error <Error>
• Rückgabe: <this>

Zerstört den Stream und emittiert optional ein 'error'-Ereignis. Nach diesem Aufruf gibt der Transform-Stream alle internen Ressourcen frei. Implementierer sollten diese Methode nicht überschreiben, sondern stattdessen readable._destroy() implementieren. Die Standardimplementierung von _destroy() für Transform emittiert ebenfalls 'close', es sei denn, emitClose ist auf false gesetzt.

Sobald destroy() aufgerufen wurde, sind weitere Aufrufe ein No-Op, und außer von _destroy() können keine weiteren Fehler als 'error' emittiert werden.

stream.duplexPair([options])#

• options <Object> Ein Wert, der an beide Duplex-Konstruktoren übergeben wird, um Optionen wie Pufferung festzulegen.
• Rückgabe: <Array> von zwei Duplex-Instanzen.

Die Hilfsfunktion duplexPair gibt ein Array mit zwei Elementen zurück, die jeweils ein mit der anderen Seite verbundener Duplex-Stream sind.

const [ sideA, sideB ] = duplexPair();
copy

Was immer in einen Stream geschrieben wird, ist auf dem anderen lesbar. Dies bietet ein Verhalten, das analog zu einer Netzwerkverbindung ist, bei der die vom Client geschriebenen Daten vom Server gelesen werden können und umgekehrt.

Die Duplex-Streams sind symmetrisch; einer oder der andere kann ohne Unterschiede im Verhalten verwendet werden.

stream.finished(stream[, options], callback)#

• stream <Stream> | <ReadableStream> | <WritableStream> Ein lesbarer und/oder schreibbarer Stream/Webstream.
• options <Object>
  • error <boolean> Wenn auf false gesetzt, wird ein Aufruf von emit('error', err) nicht als beendet behandelt. Standard: true.
  • readable <boolean> Wenn auf false gesetzt, wird der Callback aufgerufen, wenn der Stream endet, auch wenn der Stream noch lesbar sein könnte. Standard: true.
  • writable <boolean> Wenn auf false gesetzt, wird der Callback aufgerufen, wenn der Stream endet, auch wenn der Stream noch beschreibbar sein könnte. Standard: true.
  • signal <AbortSignal> ermöglicht das Abbrechen des Wartens auf das Stream-Ende. Der zugrunde liegende Stream wird nicht abgebrochen, wenn das Signal abgebrochen wird. Der Callback wird mit einem AbortError aufgerufen. Alle durch diese Funktion hinzugefügten registrierten Listener werden ebenfalls entfernt.
• callback <Function> Eine Callback-Funktion, die ein optionales Fehlerargument entgegennimmt.
• Rückgabe: <Function> Eine Bereinigungsfunktion, die alle registrierten Listener entfernt.

Eine Funktion, um benachrichtigt zu werden, wenn ein Stream nicht mehr lesbar oder beschreibbar ist oder einen Fehler oder ein vorzeitiges Close-Ereignis erfahren hat.

const { finished } = require('node:stream');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

finished(rs, (err) => {
  if (err) {
    console.error('Stream failed.', err);
  } else {
    console.log('Stream is done reading.');
  }
});

rs.resume(); // Drain the stream.
copy