Web Streams Everywhere (und Fetch für Node.js)

Chrome Developer Advocate Jake Archibald nannte 2016 „das Jahr der Web Streams.“ Seine Vorhersage war offensichtlich etwas verfrüht. Der Streams Standard wurde bereits 2014 angekündigt. Es hat eine Weile gedauert, aber es gibt jetzt eine konsistente Streaming-API, die in modernen Browsern (warte immer noch auf Firefox...) und in Node (und Deno) implementiert ist.

Was sind Streams?

Beim Streaming wird eine Ressource in kleinere Teile, sogenannte Chunks, aufgeteilt und jeder Chunk einzeln verarbeitet. Anstatt darauf warten zu müssen, den Download aller Daten abzuschließen, können Sie mit Streams Daten progressiv verarbeiten, sobald der erste Chunk verfügbar ist.

Es gibt drei Arten von Streams: Readable Streams, Writable Streams und Transform Streams. Readable Streams sind die Quelle der Daten-Chunks. Die zugrunde liegenden Datenquellen könnten zum Beispiel eine Datei oder eine HTTP-Verbindung sein. Die Daten können dann (optional) von einem Transform Stream modifiziert werden. Die Daten-Chunks können dann an einen Writable Stream weitergeleitet werden.

Web Streams überall

Node hatte schon immer seinen eigenen Typ von Streams. Diese gelten generell als schwierig zu handhaben. Der Web Streams-Standard der Web Hypertext Application Technology Working Group (WHATWG) kam später und wird weitgehend als Verbesserung angesehen. Die Node-Dokumentation nennt sie „Web Streams“, was etwas weniger umständlich klingt. Die ursprünglichen Node-Streams werden nicht veraltet oder entfernt, aber sie werden nun neben der Web-Standard-Stream-API existieren. Dies erleichtert die plattformübergreifende Codeentwicklung und bedeutet, dass Entwickler nur einen Weg erlernen müssen.

Deno, ein weiterer Versuch von Nodes ursprünglichem Erfinder, serverseitiges JavaScript zu schaffen, hat sich schon immer eng an Browser-APIs orientiert und unterstützt Web Streams vollständig. Cloudflare Workers (die Service Workern ähneln, aber an CDN-Edge-Standorten laufen) und Deno Deploy (ein Serverless-Angebot von Deno) unterstützen ebenfalls Streams.

fetch()-Antwort als Readable Stream

Es gibt mehrere Möglichkeiten, einen Readable Stream zu erstellen, aber der Aufruf von fetch() wird mit Sicherheit die häufigste sein. Der Antwortkörper von fetch() ist ein Readable Stream.

fetch('data.txt')
.then(response => console.log(response.body));

Wenn Sie in die Konsolenprotokollierung schauen, können Sie sehen, dass ein Readable Stream mehrere nützliche Methoden hat. Wie die Spezifikation besagt: „Ein Readable Stream kann direkt mit seiner pipeTo()-Methode an einen Writable Stream übergeben werden, oder er kann zuerst durch einen oder mehrere Transform Streams geleitet werden, indem seine pipeThrough()-Methode verwendet wird.“

Im Gegensatz zu Browsern implementiert Node Core derzeit kein Fetch. node-fetch, eine beliebte Abhängigkeit, die versucht, die API des Browserstandards nachzubilden, gibt einen Node Stream zurück, keinen WHATWG Stream. Undici, ein verbesserter HTTP/1.1-Client des Node.js-Teams, ist eine moderne Alternative zum Node.js-Kern http.request (auf dem Dinge wie node-fetch und Axios aufbauen). Undici hat fetch implementiert — und response.body gibt tatsächlich einen Web Stream zurück. 🎉

Undici wird möglicherweise irgendwann in den Node.js-Kern aufgenommen und wird voraussichtlich der empfohlene Weg für die Behandlung von HTTP-Anfragen in Node. Sobald Sie undici npm installieren und fetch importieren, funktioniert es genauso wie im Browser. Im folgenden Beispiel leiten wir den Stream durch einen Transform Stream. Jeder Chunk des Streams ist ein Uint8Array. Der Node-Kern stellt einen TextDecoderStream zur Dekodierung binärer Daten bereit.

import { fetch } from 'undici';
import { TextDecoderStream } from 'node:stream/web';

async function fetchStream() {
  const response = await fetch('https://example.com')
  const stream = response.body;
  const textStream = stream.pipeThrough(new TextDecoderStream());
}

response.body ist synchron, Sie müssen es also nicht awaiten. Im Browser sind fetch und TextDecoderStream im globalen Objekt verfügbar, sodass Sie keine Importanweisungen einfügen würden. Abgesehen davon ist der Code für Node und Webbrowser exakt derselbe. Deno verfügt ebenfalls über integrierte Unterstützung für fetch und TextDecoderStream.

Asynchrone Iteration

Die for-await-of-Schleife ist eine asynchrone Version der for-of-Schleife. Eine reguläre for-of-Schleife wird verwendet, um Arrays und andere iterierbare Objekte zu durchlaufen. Eine for-await-of-Schleife kann verwendet werden, um z. B. ein Array von Promises zu durchlaufen.

const promiseArray = [Promise.resolve("thing 1"), Promise.resolve("thing 2")];
for await (const thing of promiseArray) { console.log(thing); }

Wichtig für uns ist, dass dies auch zum Durchlaufen von Streams verwendet werden kann.

async function fetchAndLogStream() {
  const response = await fetch('https://example.com')
  const stream = response.body;
  const textStream = stream.pipeThrough(new TextDecoderStream());

  for await (const chunk of textStream) {
    console.log(chunk);
  }
}

fetchAndLogStream();

Die asynchrone Iteration von Streams funktioniert in Node und Deno. Alle modernen Browser haben for-await-of-Schleifen ausgeliefert, aber sie funktionieren noch nicht mit Streams.

Andere Möglichkeiten, einen Readable Stream zu erhalten

Fetch wird eine der häufigsten Möglichkeiten sein, einen Stream zu erhalten, aber es gibt noch andere. Blob und File haben beide eine .stream()-Methode, die einen Readable Stream zurückgibt. Der folgende Code funktioniert in modernen Browsern sowie in Node und Deno — in Node müssen Sie jedoch import { Blob } from 'buffer'; importieren, bevor Sie es verwenden können.

const blobStream = new Blob(['Lorem ipsum'], { type: 'text/plain' }).stream();

Hier ist ein Beispiel für einen Frontend-Browser: Wenn Sie ein <input type="file"> in Ihrem Markup haben, ist es einfach, die vom Benutzer ausgewählte Datei als Stream zu erhalten.

const fileStream = document.querySelector('input').files[0].stream();

In Node 17 wurde das FileHandle-Objekt, das von der fs/promises open()-Funktion zurückgegeben wird, um die Methode .readableWebStream() erweitert.

import {
  open,
} from 'node:fs/promises';

const file = await open('./some/file/to/read');

for await (const chunk of file.readableWebStream())
  console.log(chunk);

await file.close();

Streams arbeiten gut mit Promises zusammen

Wenn Sie etwas tun müssen, nachdem der Stream abgeschlossen ist, können Sie Promises verwenden.

someReadableStream
.pipeTo(someWritableStream)
.then(() => console.log("all data successfully written"))
.catch(error => console.error("something went wrong", error))

Alternativ können Sie das Ergebnis auch optional awaiten.

await someReadableStream.pipeTo(someWritableStream)

Erstellen Ihres eigenen Transform Streams

Wir haben TextDecoderStream bereits gesehen (es gibt auch einen TextEncoderStream). Sie können auch Ihren eigenen Transform Stream von Grund auf neu erstellen. Der Konstruktor TransformStream kann ein Objekt akzeptieren. Sie können drei Methoden im Objekt angeben: start, transform und flush. Diese sind alle optional, aber transform ist das, was die Transformation tatsächlich durchführt.

Als Beispiel tun wir so, als ob TextDecoderStream() nicht existiert, und implementieren die gleiche Funktionalität (verwenden Sie in der Produktion unbedingt TextDecoderStream, da das Folgende ein stark vereinfachtes Beispiel ist).

const decoder = new TextDecoder();
const decodeStream = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(decoder.decode(chunk, {stream: true}));
  }
});

Jeder empfangene Chunk wird modifiziert und dann vom Controller weitergeleitet. Im obigen Beispiel ist jeder Chunk eine Art kodierter Text, der dekodiert und dann weitergeleitet wird. Werfen wir einen kurzen Blick auf die beiden anderen Methoden.

const transformStream = new TransformStream({
  start(controller) {
    // Called immediately when the TransformStream is created
  },

  flush(controller) {
    // Called when chunks are no longer being forwarded to the transformer
  }
});

Ein Transform Stream ist ein Readable Stream und ein Writable Stream, die zusammenarbeiten, normalerweise um Daten zu transformieren. Jedes Objekt, das mit new TransformStream() erstellt wird, hat eine Eigenschaft namens readable, die ein ReadableStream ist, und eine Eigenschaft namens writable, die ein Writable Stream ist. Der Aufruf von someReadableStream.pipeThrough() schreibt die Daten von someReadableStream in transformStream.writable, transformiert die Daten möglicherweise und leitet die Daten dann an transformStream.readable weiter.

Manche Leute finden es hilfreich, einen Transform Stream zu erstellen, der die Daten nicht tatsächlich transformiert. Dies wird als „Identity Transform Stream“ bezeichnet — erstellt durch den Aufruf von new TransformStream() ohne Übergabe eines Objektarguments oder durch Weglassen der transform-Methode. Er leitet alle Chunks, die an seine beschreibbare Seite geschrieben werden, unverändert an seine lesbare Seite weiter. Als einfaches Beispiel für das Konzept wird „hello“ vom folgenden Code protokolliert.

const {readable, writable} = new TransformStream();
writable.getWriter().write('hello');
readable.getReader().read().then(({value, done}) => console.log(value))

Erstellen Ihres eigenen Readable Streams

Es ist möglich, einen benutzerdefinierten Stream zu erstellen und ihn mit eigenen Chunks zu befüllen. Der Konstruktor new ReadableStream() nimmt ein Objekt entgegen, das eine start-Funktion, eine pull-Funktion und eine cancel-Funktion enthalten kann. Diese Funktion wird sofort aufgerufen, wenn der ReadableStream erstellt wird. Verwenden Sie innerhalb der start-Funktion controller.enqueue, um Chunks zum Stream hinzuzufügen.

Hier ist ein einfaches „Hallo Welt“-Beispiel.

import { ReadableStream } from "node:stream/web";
const readable = new ReadableStream({
  start(controller) {
    controller.enqueue("hello");
    controller.enqueue("world");
    controller.close();
  },
});

const allChunks = [];
for await (const chunk of readable) {
  allChunks.push(chunk);
}
console.log(allChunks.join(" "));

Hier ist ein praxisnäheres Beispiel aus der Streams-Spezifikation, das einen Websocket in einen Readable Stream umwandelt.

function makeReadableWebSocketStream(url, protocols) {
  let websocket = new WebSocket(url, protocols);
  websocket.binaryType = "arraybuffer";

  return new ReadableStream({
    start(controller) {
      websocket.onmessage = event => controller.enqueue(event.data);
      websocket.onclose = () => controller.close();
      websocket.onerror = () => controller.error(new Error("The WebSocket errored"));
    }
  });
}

Interoperabilität von Node Streams

In Node wird die alte, Node-spezifische Art der Arbeit mit Streams nicht entfernt. Die alte Node-Streams-API und die Web-Streams-API werden nebeneinander existieren. Es kann daher manchmal notwendig sein, einen Node-Stream in einen Web-Stream und umgekehrt umzuwandeln, indem die Methoden .fromWeb() und .toWeb() verwendet werden, die in Node 17 hinzugefügt werden.

import {Readable} from 'node:stream';
import {fetch} from 'undici';

const response = await fetch(url);
const readableNodeStream = Readable.fromWeb(response.body);

Fazit

ES-Module, EventTarget, AbortController, URL-Parser, Web Crypto, Blob, TextEncoder/Decoder: Immer mehr Browser-APIs finden ihren Weg in Node.js. Das Wissen und die Fähigkeiten sind übertragbar. Fetch und Streams sind ein wichtiger Teil dieser Konvergenz.

Domenic Denicola, Mitherausgeber der Streams-Spezifikation, hat geschrieben, dass das Ziel der Streams-API darin besteht, eine effiziente Abstraktion und ein vereinheitlichendes Primitiv für E/A bereitzustellen, so wie Promises es für Asynchronität geworden sind. Um im Frontend wirklich nützlich zu sein, müssen mehr APIs Streams unterstützen. Derzeit ist ein MediaStream trotz seines Namens kein Readable Stream. Wenn Sie mit Video oder Audio arbeiten (zumindest im Moment), kann ein Readable Stream nicht srcObject zugewiesen werden. Oder nehmen wir an, Sie möchten ein Bild abrufen, es durch einen Transform Stream leiten und es dann auf die Seite einfügen. Zum Zeitpunkt des Schreibens ist der Code für die Verwendung eines Streams als src eines Bildelements etwas umständlich.

const response = await fetch('cute-cat.png');
const bodyStream = response.body;
const newResponse = new Response(bodyStream);
const blob = await newResponse.blob();
const url = URL.createObjectURL(blob);
document.querySelector('img').src = url;    

Mit der Zeit werden jedoch mehr APIs sowohl im Browser als auch in Node (und Deno) Streams nutzen, daher lohnt es sich, sich damit zu beschäftigen. Es gibt bereits eine Stream-API für die Arbeit mit WebSockets in Deno und Chrome. Chrome hat Fetch-Anfrages-Streams implementiert. Node und Chrome haben übertragbare Streams implementiert, um Daten an und von einem Worker zu leiten, um die Chunks in einem separaten Thread zu verarbeiten. Leute nutzen bereits Streams, um interessante Dinge für Produkte in der realen Welt zu tun: Die Entwickler der Dateifreigabe-Web-App Wormhole haben Code zum Verschlüsseln eines Streams Open-Source gestellt.

Vielleicht wird 2022 das Jahr der Web Streams...