Streams – przewodnik

Dowiedz się, jak używać strumieni do odczytu, zapisu i przekształcania za pomocą interfejsu Streams API.

Thomas Steiner

Interfejs Streams API umożliwia programowy dostęp do strumieni danych otrzymywanych przez sieć lub tworzonych lokalnie w dowolny sposób oraz przetwarzanie ich za pomocą JavaScriptu. Strumieniowanie polega na podzieleniu zasobu, który chcesz otrzymać, wysłać lub przekształcić, na małe części, a następnie przetwarzaniu ich po kolei. Przeglądarki i tak przesyłają strumieniowo dane, gdy otrzymują komponenty, takie jak HTML czy filmy, które mają być wyświetlane na stronach internetowych. Jednak ta funkcja nigdy wcześniej nie była dostępna w JavaScript. Zmieniło się to w 2015 r., gdy wprowadzono fetch.

Uwaga Streaming jest technicznie możliwy w przypadku XMLHttpRequest, ale nie jest to optymalne rozwiązanie. Oto gist na GitHubie z dobrym przykładem użycia „XMLHttpRequest”.

Wcześniej, jeśli chciałeś przetworzyć jakiś zasób (np. plik wideo lub tekstowy), musiałeś pobrać cały plik, poczekać, aż zostanie zdeserializowany do odpowiedniego formatu, a następnie go przetworzyć. Dzięki temu, że strumienie są dostępne w JavaScript, wszystko się zmienia. Możesz teraz przetwarzać nieprzetworzone dane za pomocą JavaScriptu w miarę ich dostępności na urządzeniu klienta, bez konieczności generowania bufora, ciągu znaków ani obiektu blob. Otwiera to wiele możliwości, z których niektóre wymieniam poniżej:

• Efekty wideo: przesyłanie czytelnego strumienia wideo przez strumień przekształcający, który stosuje efekty w czasie rzeczywistym.
• Kompresja i dekompresja danych: przesyłanie strumienia plików przez strumień przekształcania, który selektywnie kompresuje lub dekompresuje dane.
• Dekodowanie obrazu: przekazywanie strumienia odpowiedzi HTTP przez strumień przekształcający, który dekoduje bajty na dane bitmapy, a następnie przez kolejny strumień przekształcający, który tłumaczy bitmapy na pliki PNG. Jeśli jest zainstalowany w obsłudze zdarzenia fetch w usłudze Service Worker, umożliwia przezroczyste uzupełnianie nowych formatów obrazów, takich jak AVIF.

Obsługa przeglądarek

ReadableStream i WritableStream

TransformStream

Podstawowe pojęcia

Zanim przejdę do szczegółów dotyczących różnych typów strumieni, przedstawię kilka podstawowych pojęć.

Grudki

Fragment to pojedynczy element danych zapisywany w strumieniu lub z niego odczytywany. Może to być dowolny typ. Strumienie mogą nawet zawierać fragmenty różnych typów. Zwykle fragment nie jest najmniejszą jednostką danych w danym strumieniu. Na przykład strumień bajtów może zawierać fragmenty składające się z 16 KiB Uint8Array zamiast pojedynczych bajtów.

Strumienie do odczytu

Strumień odczytu reprezentuje źródło danych, z którego można odczytywać informacje. Innymi słowy, dane wypływają ze strumienia odczytu. Strumień do odczytu to instancja klasy ReadableStream.

Strumienie z możliwością zapisu

Strumień z możliwością zapisu to miejsce docelowe danych, w którym możesz zapisywać informacje. Inaczej mówiąc, dane trafiają do strumienia z możliwością zapisu. Strumień zapisywalny to instancja klasy WritableStream.

Przekształcanie strumieni

Strumień przekształcający składa się z pary strumieni: strumienia do zapisu, zwanego stroną do zapisu, i strumienia do odczytu, zwanego stroną do odczytu. W realnym świecie można to porównać do tłumacza symultanicznego, który tłumaczy z jednego języka na drugi na bieżąco. W przypadku strumienia przekształcania zapisywanie po stronie zapisu powoduje udostępnienie nowych danych do odczytu po stronie odczytu. Każdy obiekt z właściwościami writable i readable może służyć jako strumień przekształceń. Standardowa klasa TransformStream ułatwia jednak tworzenie takich par, które są prawidłowo powiązane.

Łańcuchy do rur

Strumienie są używane głównie przez przekazywanie ich do siebie. Strumień do odczytu można przekierować bezpośrednio do strumienia do zapisu za pomocą metody pipeTo() strumienia do odczytu lub przekierować najpierw przez co najmniej 1 strumień przekształcający za pomocą metody pipeThrough() strumienia do odczytu. Zestaw połączonych strumieni jest nazywany łańcuchem potoków.

Ciśnienie zwrotne

Po utworzeniu łańcucha potoków będzie on przekazywać sygnały dotyczące tego, jak szybko powinny przez niego przepływać fragmenty. Jeśli którykolwiek etap w łańcuchu nie może jeszcze przyjmować fragmentów, przesyła sygnał wstecz przez łańcuch potokowy, aż do momentu, gdy pierwotne źródło otrzyma polecenie, aby nie produkować fragmentów tak szybko. Ten proces normalizacji przepływu jest nazywany wstecznym ciśnieniem.

Teeing

Strumień odczytu można rozdzielić (nazwa pochodzi od kształtu wielkiej litery „T”) za pomocą metody tee(). Spowoduje to zablokowanie strumienia, czyli uniemożliwi jego bezpośrednie użycie. Utworzy jednak 2 nowe strumienie, zwane gałęziami, które można wykorzystywać niezależnie. Teeing jest też ważny, ponieważ strumieni nie można przewijać ani ponownie uruchamiać (więcej informacji na ten temat znajdziesz w dalszej części artykułu).

Mechanizm strumienia odczytu

Strumień do odczytu to źródło danych reprezentowane w JavaScript przez obiekt ReadableStream, który przepływa z bazowego źródła. Konstruktor ReadableStream() tworzy i zwraca obiekt strumienia do odczytu na podstawie podanych funkcji obsługi. Istnieją 2 typy źródeł danych:

• Źródła typu push stale przesyłają dane, gdy uzyskasz do nich dostęp. To Ty decydujesz, czy chcesz rozpocząć, wstrzymać lub anulować dostęp do strumienia. Przykłady to strumienie wideo na żywo, zdarzenia wysyłane przez serwer lub WebSockety.
• Źródła pobierania wymagają, aby po połączeniu z nimi wyraźnie poprosić o dane. Przykłady obejmują operacje HTTP za pomocą wywołań fetch() lub XMLHttpRequest.

Dane strumieniowe są odczytywane sekwencyjnie w małych częściach zwanych fragmentami. Fragmenty umieszczane w strumieniu są enqueued. Oznacza to, że czekają w kolejce na odczytanie. Kolejka wewnętrzna śledzi fragmenty, które nie zostały jeszcze odczytane.

Strategia kolejkowania to obiekt, który określa, jak strumień powinien sygnalizować nadmierne obciążenie na podstawie stanu wewnętrznej kolejki. Strategia kolejkowania przypisuje rozmiar do każdego fragmentu i porównuje łączny rozmiar wszystkich fragmentów w kolejce z określoną liczbą, zwaną wysokim poziomem.

Fragmenty w strumieniu są odczytywane przez czytnik. Ten czytnik pobiera dane po jednym bloku, co pozwala wykonywać na nich dowolne operacje. Czytnik wraz z kodem przetwarzania nazywa się konsumentem.

Kolejna konstrukcja w tym kontekście to kontroler. Każdy strumień do odczytu ma powiązany kontroler, który, jak sama nazwa wskazuje, umożliwia sterowanie strumieniem.

W danym momencie strumień może odczytywać tylko 1 czytelnik. Gdy czytelnik zostanie utworzony i zacznie odczytywać strumień (czyli stanie się aktywnym czytelnikiem), zostanie z nim zablokowany. Jeśli chcesz, aby inny czytnik przejął odczytywanie strumienia, zwykle musisz zwolnić pierwszy czytnik, zanim zrobisz cokolwiek innego (chociaż możesz rozdzielić strumienie).

Tworzenie strumienia do odczytu

Strumień do odczytu tworzy się przez wywołanie jego konstruktora ReadableStream(). Konstruktor ma argument opcjonalny underlyingSource, który reprezentuje obiekt z metodami i właściwościami określającymi sposób działania utworzonej instancji strumienia.

underlyingSource

Może to wykorzystywać te opcjonalne metody zdefiniowane przez dewelopera:

• start(controller): wywoływana natychmiast po utworzeniu obiektu. Metoda może uzyskać dostęp do źródła strumienia i wykonać wszystkie inne czynności wymagane do skonfigurowania funkcji strumienia. Jeśli ten proces ma być wykonywany asynchronicznie, metoda może zwrócić obietnicę, aby zasygnalizować powodzenie lub niepowodzenie. Parametr controller przekazywany do tej metody to ReadableStreamDefaultController.
• pull(controller): może służyć do sterowania strumieniem w miarę pobierania kolejnych fragmentów. Jest ona wywoływana wielokrotnie, dopóki wewnętrzna kolejka fragmentów strumienia nie jest pełna, aż do osiągnięcia przez nią wysokiego poziomu. Jeśli wynikiem wywołania funkcji pull() jest obietnica, funkcja pull() nie zostanie ponownie wywołana, dopóki ta obietnica nie zostanie spełniona. Jeśli obietnica zostanie odrzucona, strumień będzie zawierać błędy.
• cancel(reason): wywoływana, gdy odbiorca strumienia anuluje strumień.

const readableStream = new ReadableStream({
  start(controller) {
    /* … */
  },

  pull(controller) {
    /* … */
  },

  cancel(reason) {
    /* … */
  },
});

ReadableStreamDefaultController obsługuje te metody:

• ReadableStreamDefaultController.close() zamyka powiązany strumień.
• ReadableStreamDefaultController.enqueue() umieszcza dany fragment w powiązanym strumieniu.
• ReadableStreamDefaultController.error() powoduje, że wszystkie przyszłe interakcje z powiązanym strumieniem będą generować błędy.

/* … */
start(controller) {
  controller.enqueue('The first chunk!');
},
/* … */

queuingStrategy

Drugi, również opcjonalny, argument konstruktora ReadableStream() to queuingStrategy. Jest to obiekt, który opcjonalnie definiuje strategię kolejkowania strumienia. Przyjmuje 2 parametry:

• highWaterMark: nieujemna liczba wskazująca górny próg strumienia korzystającego z tej strategii kolejkowania.
• size(chunk): funkcja, która oblicza i zwraca skończony nieujemny rozmiar podanej wartości fragmentu. Wynik jest używany do określania ciśnienia zwrotnego, które objawia się za pomocą odpowiedniej właściwości ReadableStreamDefaultController.desiredSize. Określa też, kiedy wywoływana jest metoda pull() źródła bazowego.

const readableStream = new ReadableStream({
    /* … */
  },
  {
    highWaterMark: 10,
    size(chunk) {
      return chunk.length;
    },
  },
);

Metody getReader() i read()

Aby odczytać dane ze strumienia do odczytu, potrzebujesz czytnika, który będzie obiektem ReadableStreamDefaultReader. Metoda getReader() interfejsu ReadableStream tworzy czytnik i blokuje strumień. Gdy strumień jest zablokowany, nie można uzyskać dostępu do innego czytnika, dopóki ten nie zostanie zwolniony.

Metoda read() interfejsu ReadableStreamDefaultReader zwraca obietnicę zapewniającą dostęp do następnego fragmentu w wewnętrznej kolejce strumienia. Zwraca wynik lub odrzuca go w zależności od stanu strumienia. Możliwe są te opcje:

• Jeśli fragment jest dostępny, obietnica zostanie spełniona za pomocą obiektu w formacie
  { value: chunk, done: false }.
• Jeśli strumień zostanie zamknięty, obietnica zostanie spełniona za pomocą obiektu w formacie
  { value: undefined, done: true }.
• Jeśli strumień zwróci błąd, obietnica zostanie odrzucona z odpowiednim błędem.

const reader = readableStream.getReader();
while (true) {
  const { done, value } = await reader.read();
  if (done) {
    console.log('The stream is done.');
    break;
  }
  console.log('Just read a chunk:', value);
}

Usługa locked

Możesz sprawdzić, czy strumień do odczytu jest zablokowany, uzyskując dostęp do jego właściwości ReadableStream.locked.

const locked = readableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

Przykładowy kod strumienia do odczytu

Poniższy przykładowy kod pokazuje wszystkie kroki w działaniu. Najpierw utwórz ReadableStream, który w argumencie underlyingSource (czyli w klasie TimestampSource) definiuje metodę start(). Ta metoda informuje strumień controller o enqueue() sygnaturze czasowej co sekundę przez 10 sekund. Na koniec informuje kontroler, aby close() strumień. Aby korzystać z tego strumienia, utwórz czytnik za pomocą metody getReader() i wywołuj read(), dopóki strumień nie zostanie done.

class TimestampSource {
  #interval

  start(controller) {
    this.#interval = setInterval(() => {
      const string = new Date().toLocaleTimeString();
      // Add the string to the stream.
      controller.enqueue(string);
      console.log(`Enqueued ${string}`);
    }, 1_000);

    setTimeout(() => {
      clearInterval(this.#interval);
      // Close the stream after 10s.
      controller.close();
    }, 10_000);
  }

  cancel() {
    // This is called if the reader cancels.
    clearInterval(this.#interval);
  }
}

const stream = new ReadableStream(new TimestampSource());

async function concatStringStream(stream) {
  let result = '';
  const reader = stream.getReader();
  while (true) {
    // The `read()` method returns a promise that
    // resolves when a value has been received.
    const { done, value } = await reader.read();
    // Result objects contain two properties:
    // `done`  - `true` if the stream has already given you all its data.
    // `value` - Some data. Always `undefined` when `done` is `true`.
    if (done) return result;
    result += value;
    console.log(`Read ${result.length} characters so far`);
    console.log(`Most recently read chunk: ${value}`);
  }
}
concatStringStream(stream).then((result) => console.log('Stream complete', result));

Iteracja asynchroniczna

Sprawdzanie, czy strumień jest done, przy każdej iteracji pętli read() może nie być najwygodniejszym interfejsem API. Na szczęście wkrótce pojawi się lepszy sposób na to: iteracja asynchroniczna.

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

Obecnie obejściem umożliwiającym korzystanie z iteracji asynchronicznej jest zaimplementowanie tego działania za pomocą polyfill.

if (!ReadableStream.prototype[Symbol.asyncIterator]) {
  ReadableStream.prototype[Symbol.asyncIterator] = async function* () {
    const reader = this.getReader();
    try {
      while (true) {
        const {done, value} = await reader.read();
        if (done) {
          return;
          }
        yield value;
      }
    }
    finally {
      reader.releaseLock();
    }
  }
}

Rozdzielanie strumienia do odczytu

Metoda tee() interfejsu ReadableStream dzieli bieżący strumień do odczytu, zwracając dwuelementową tablicę zawierającą 2 wynikowe gałęzie jako nowe instancje ReadableStream. Umożliwia to dwóm czytnikom jednoczesne odczytywanie strumienia. Możesz to zrobić na przykład w skrypcie service worker, jeśli chcesz pobrać odpowiedź z serwera i przesłać ją strumieniowo do przeglądarki, ale też do pamięci podręcznej skryptu service worker. Treści odpowiedzi nie można użyć więcej niż raz, więc potrzebujesz 2 kopii. Aby anulować strumień, musisz anulować obie powstałe gałęzie. Utworzenie kopii strumienia zwykle blokuje go na cały czas trwania, uniemożliwiając innym czytnikom zablokowanie go.

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called `read()` when the controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

// Create two `ReadableStream`s.
const [streamA, streamB] = readableStream.tee();

// Read streamA iteratively one by one. Typically, you
// would not do it this way, but you certainly can.
const readerA = streamA.getReader();
console.log('[A]', await readerA.read()); //=> {value: "a", done: false}
console.log('[A]', await readerA.read()); //=> {value: "b", done: false}
console.log('[A]', await readerA.read()); //=> {value: "c", done: false}
console.log('[A]', await readerA.read()); //=> {value: "d", done: false}
console.log('[A]', await readerA.read()); //=> {value: undefined, done: true}

// Read streamB in a loop. This is the more common way
// to read data from the stream.
const readerB = streamB.getReader();
while (true) {
  const result = await readerB.read();
  if (result.done) break;
  console.log('[B]', result);
}

Strumienie bajtów do odczytu

W przypadku strumieni reprezentujących bajty udostępniana jest rozszerzona wersja strumienia do odczytu, która umożliwia wydajne przetwarzanie bajtów, w szczególności przez minimalizowanie kopii. Strumienie bajtów umożliwiają pozyskiwanie czytelników, którzy korzystają z własnego bufora (BYOB). Domyślna implementacja może dawać różne dane wyjściowe, np. ciągi znaków lub bufory tablicowe w przypadku WebSocketów, natomiast strumienie bajtów gwarantują dane wyjściowe w postaci bajtów. Czytniki BYOB mają też zalety związane ze stabilnością. Dzieje się tak, ponieważ jeśli bufor zostanie odłączony, można zagwarantować, że nie będzie można zapisać w tym samym buforze dwa razy, co pozwala uniknąć sytuacji wyścigu. Czytniki BYOB mogą zmniejszyć liczbę uruchomień odśmiecania pamięci w przeglądarce, ponieważ mogą ponownie wykorzystywać bufory.

Tworzenie strumienia bajtów do odczytu

Możesz utworzyć czytelny strumień bajtów, przekazując dodatkowy parametr type do konstruktora ReadableStream().

new ReadableStream({ type: 'bytes' });

underlyingSource

Podstawowe źródło czytelnego strumienia bajtów otrzymuje ReadableByteStreamController do manipulowania. Jej metoda ReadableByteStreamController.enqueue() przyjmuje argument chunk, którego wartością jest ArrayBufferView. Właściwość ReadableByteStreamController.byobRequest zwraca bieżące żądanie pull BYOB lub wartość null, jeśli nie ma takiego żądania. Na koniec właściwość ReadableByteStreamController.desiredSize zwraca żądany rozmiar, aby wypełnić wewnętrzną kolejkę kontrolowanego strumienia.

queuingStrategy

Drugi, również opcjonalny, argument konstruktora ReadableStream() to queuingStrategy. Jest to obiekt, który opcjonalnie definiuje strategię kolejkowania strumienia. Przyjmuje on jeden parametr:

• highWaterMark: nieujemna liczba bajtów wskazująca maksymalną ilość danych w strumieniu korzystającym z tej strategii kolejkowania. Służy to do określania nadmiernego obciążenia, które objawia się za pomocą odpowiedniej właściwości ReadableByteStreamController.desiredSize. Określa też, kiedy wywoływana jest metoda pull() źródła bazowego.

Metody getReader() i read()

Następnie możesz uzyskać dostęp do ReadableStreamBYOBReader, odpowiednio ustawiając parametr mode: ReadableStream.getReader({ mode: "byob" }). Umożliwia to precyzyjniejszą kontrolę nad przydzielaniem bufora, aby uniknąć kopii. Aby odczytać dane ze strumienia bajtów, musisz wywołać funkcję ReadableStreamBYOBReader.read(view), gdzie view jest obiektem ArrayBufferView.

Przykładowy kod czytelnego strumienia bajtów

const reader = readableStream.getReader({ mode: "byob" });

let startingAB = new ArrayBuffer(1_024);
const buffer = await readInto(startingAB);
console.log("The first 1024 bytes, or less:", buffer);

async function readInto(buffer) {
  let offset = 0;

  while (offset < buffer.byteLength) {
    const { value: view, done } =
        await reader.read(new Uint8Array(buffer, offset, buffer.byteLength - offset));
    buffer = view.buffer;
    if (done) {
      break;
    }
    offset += view.byteLength;
  }

  return buffer;
}

Ta funkcja zwraca czytelne strumienie bajtów, które umożliwiają wydajne odczytywanie bez kopiowania losowo wygenerowanej tablicy. Zamiast używać z góry określonego rozmiaru fragmentu wynoszącego 1024 bajty, próbuje wypełnić bufor dostarczony przez dewelopera, co zapewnia pełną kontrolę.

const DEFAULT_CHUNK_SIZE = 1_024;

function makeReadableByteStream() {
  return new ReadableStream({
    type: 'bytes',

    pull(controller) {
      // Even when the consumer is using the default reader,
      // the auto-allocation feature allocates a buffer and
      // passes it to us via `byobRequest`.
      const view = controller.byobRequest.view;
      view = crypto.getRandomValues(view);
      controller.byobRequest.respond(view.byteLength);
    },

    autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
  });
}

Mechanizm strumienia zapisu

Strumień zapisu to miejsce docelowe, do którego możesz zapisywać dane. W JavaScript jest on reprezentowany przez obiekt WritableStream. Jest to abstrakcja podstawowego ujścia, czyli ujścia wejścia/wyjścia niższego poziomu, do którego zapisywane są dane pierwotne.

Dane są zapisywane w strumieniu za pomocą zapisywacza, po jednym bloku naraz. Fragment może przybierać wiele form, podobnie jak fragmenty w czytniku. Możesz użyć dowolnego kodu, aby przygotować fragmenty gotowe do napisania. Osoba pisząca wraz z powiązanym kodem to producent.

Gdy moduł zapisujący zostanie utworzony i zacznie zapisywać dane w strumieniu (aktywny moduł zapisujący), mówi się, że jest on zablokowany. W danym momencie tylko jeden zapisujący może zapisywać dane w strumieniu z możliwością zapisu. Jeśli chcesz, aby inny proces zapisu zaczął zapisywać dane do strumienia, musisz zwykle zwolnić strumień, a następnie dołączyć do niego inny proces zapisu.

Kolejka wewnętrzna śledzi fragmenty, które zostały zapisane w strumieniu, ale nie zostały jeszcze przetworzone przez bazowy odbiornik.

Strategia kolejkowania to obiekt, który określa, jak strumień powinien sygnalizować nadmierne obciążenie na podstawie stanu wewnętrznej kolejki. Strategia kolejkowania przypisuje rozmiar do każdego fragmentu i porównuje łączny rozmiar wszystkich fragmentów w kolejce z określoną liczbą, zwaną wysokim poziomem.

Ostateczna konstrukcja jest nazywana kontrolerem. Każdy strumień z możliwością zapisu ma powiązany kontroler, który umożliwia sterowanie strumieniem (np. przerwanie go).

Tworzenie strumienia zapisu

Interfejs WritableStream interfejsu Streams API zapewnia standardową abstrakcję do zapisywania danych strumieniowych w miejscu docelowym, zwanym ujściem. Ten obiekt ma wbudowane funkcje ciśnienia zwrotnego i kolejkowania. Strumień do zapisu tworzy się przez wywołanie jego konstruktora WritableStream(). Ma opcjonalny parametr underlyingSink, który reprezentuje obiekt z metodami i właściwościami określającymi sposób działania utworzonej instancji strumienia.

underlyingSink

underlyingSink może zawierać te opcjonalne metody zdefiniowane przez dewelopera: Parametr controllerprzekazywany do niektórych metod jest wartością WritableStreamDefaultController.

• start(controller): ta metoda jest wywoływana natychmiast po utworzeniu obiektu. Treść tej metody powinna umożliwiać uzyskanie dostępu do bazowego miejsca docelowego. Jeśli ten proces ma być wykonywany asynchronicznie, może zwrócić obietnicę, aby zasygnalizować powodzenie lub niepowodzenie.
• write(chunk, controller): ta metoda jest wywoływana, gdy nowy fragment danych (określony w parametrze chunk) jest gotowy do zapisania w bazowym odbiorniku. Może zwrócić obietnicę, aby zasygnalizować powodzenie lub niepowodzenie operacji zapisu. Ta metoda będzie wywoływana tylko po pomyślnym zakończeniu poprzednich operacji zapisu i nigdy po zamknięciu lub przerwaniu strumienia.
• close(controller): ta metoda zostanie wywołana, jeśli aplikacja zasygnalizuje, że zakończyła zapisywanie fragmentów w strumieniu. Zawartość powinna wykonać wszystkie niezbędne czynności, aby zakończyć zapisywanie w podstawowym miejscu docelowym i zwolnić do niego dostęp. Jeśli ten proces jest asynchroniczny, może zwrócić obietnicę, aby zasygnalizować powodzenie lub niepowodzenie. Ta metoda zostanie wywołana dopiero po pomyślnym zakończeniu wszystkich zapisów w kolejce.
• abort(reason): ta metoda zostanie wywołana, jeśli aplikacja zasygnalizuje, że chce nagle zamknąć strumień i przekształcić go w stan błędu. Może ona wyczyścić wszystkie zajęte zasoby, podobnie jak funkcja close(), ale funkcja abort() zostanie wywołana nawet wtedy, gdy zapisy są w kolejce. Te fragmenty zostaną odrzucone. Jeśli ten proces jest asynchroniczny, może zwrócić obietnicę, aby zasygnalizować powodzenie lub niepowodzenie. Parametr reason zawiera DOMString z wyjaśnieniem, dlaczego strumień został przerwany.

const writableStream = new WritableStream({
  start(controller) {
    /* … */
  },

  write(chunk, controller) {
    /* … */
  },

  close(controller) {
    /* … */
  },

  abort(reason) {
    /* … */
  },
});

Interfejs WritableStreamDefaultController interfejsu Streams API reprezentuje kontroler, który umożliwia kontrolowanie stanu WritableStream podczas konfiguracji, w miarę przesyłania kolejnych fragmentów do zapisu lub na końcu zapisu. Podczas tworzenia WritableStream podstawowy odbiornik otrzymuje odpowiednią instancję WritableStreamDefaultController, którą może manipulować. WritableStreamDefaultController ma tylko jedną metodę:WritableStreamDefaultController.error(), która powoduje, że wszystkie przyszłe interakcje z powiązaną transmisją będą generować błędy. WritableStreamDefaultController obsługuje też właściwość signal, która zwraca instancję AbortSignal, umożliwiając w razie potrzeby zatrzymanie operacji WritableStream.

/* … */
write(chunk, controller) {
  try {
    // Try to do something dangerous with `chunk`.
  } catch (error) {
    controller.error(error.message);
  }
},
/* … */

queuingStrategy

Drugi, również opcjonalny, argument konstruktora WritableStream() to queuingStrategy. Jest to obiekt, który opcjonalnie definiuje strategię kolejkowania strumienia. Przyjmuje 2 parametry:

• highWaterMark: nieujemna liczba wskazująca górny próg strumienia korzystającego z tej strategii kolejkowania.
• size(chunk): funkcja, która oblicza i zwraca skończony nieujemny rozmiar podanej wartości fragmentu. Wynik jest używany do określania ciśnienia zwrotnego, które objawia się za pomocą odpowiedniej właściwości WritableStreamDefaultWriter.desiredSize.

Metody getWriter() i write()

Aby pisać w strumieniu z możliwością zapisu, potrzebujesz narzędzia do pisania, które będzieWritableStreamDefaultWriter. Metoda getWriter() interfejsu WritableStream zwraca nową instancję WritableStreamDefaultWriter i blokuje strumień w tej instancji. Gdy strumień jest zablokowany, nie można uzyskać dostępu do innego modułu zapisującego, dopóki bieżący nie zostanie zwolniony.

Metoda write() interfejsu WritableStreamDefaultWriter zapisuje przekazany fragment danych w obiekcie WritableStream i jego bazowym ujściu, a następnie zwraca obietnicę, która jest rozwiązywana, aby wskazać powodzenie lub niepowodzenie operacji zapisu. Pamiętaj, że to, co oznacza „sukces”, zależy od źródła danych. Może to oznaczać, że fragment został zaakceptowany, ale niekoniecznie, że został bezpiecznie zapisany w miejscu docelowym.

const writer = writableStream.getWriter();
const resultPromise = writer.write('The first chunk!');

Usługa locked

Możesz sprawdzić, czy strumień zapisu jest zablokowany, uzyskując dostęp do jego właściwości WritableStream.locked.

const locked = writableStream.locked;
console.log(`The stream is ${locked ? 'indeed' : 'not'} locked.`);

Przykładowy kod strumienia zapisu

Poniższy przykładowy kod pokazuje wszystkie kroki w działaniu.

const writableStream = new WritableStream({
  start(controller) {
    console.log('[start]');
  },
  async write(chunk, controller) {
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

const writer = writableStream.getWriter();
const start = Date.now();
for (const char of 'abcdefghijklmnopqrstuvwxyz') {
  // Wait to add to the write queue.
  await writer.ready;
  console.log('[ready]', Date.now() - start, 'ms');
  // The Promise is resolved after the write finishes.
  writer.write(char);
}
await writer.close();

Przekazywanie strumienia do odczytu do strumienia do zapisu

Strumień do odczytu można przekierować do strumienia do zapisu za pomocą metody pipeTo() strumienia do odczytu. ReadableStream.pipeTo() przekazuje bieżący ReadableStream do danego WritableStream i zwraca obietnicę, która zostanie spełniona, gdy proces przekazywania zakończy się pomyślnie, lub odrzucona, jeśli wystąpią błędy.

const readableStream = new ReadableStream({
  start(controller) {
    // Called by constructor.
    console.log('[start readable]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // Called when controller's queue is empty.
    console.log('[pull]');
    controller.enqueue('d');
    controller.close();
  },
  cancel(reason) {
    // Called when the stream is canceled.
    console.log('[cancel]', reason);
  },
});

const writableStream = new WritableStream({
  start(controller) {
    // Called by constructor
    console.log('[start writable]');
  },
  async write(chunk, controller) {
    // Called upon writer.write()
    console.log('[write]', chunk);
    // Wait for next write.
    await new Promise((resolve) => setTimeout(() => {
      document.body.textContent += chunk;
      resolve();
    }, 1_000));
  },
  close(controller) {
    console.log('[close]');
  },
  abort(reason) {
    console.log('[abort]', reason);
  },
});

await readableStream.pipeTo(writableStream);
console.log('[finished]');

Tworzenie strumienia przekształceń

Interfejs TransformStream interfejsu Streams API reprezentuje zestaw danych, które można przekształcać. Strumień przekształcający tworzy się przez wywołanie jego konstruktora TransformStream(), który tworzy i zwraca obiekt strumienia przekształcającego na podstawie podanych funkcji obsługi. Konstruktor TransformStream() przyjmuje jako pierwszy argument opcjonalny obiekt JavaScript reprezentujący transformer. Takie obiekty mogą zawierać dowolną z tych metod:

transformer

• start(controller): ta metoda jest wywoływana natychmiast po utworzeniu obiektu. Zwykle służy do umieszczania w kolejce fragmentów prefiksu za pomocą controller.enqueue(). Te fragmenty będą odczytywane ze strony do odczytu, ale nie zależą od żadnych zapisów na stronie do zapisu. Jeśli ten początkowy proces jest asynchroniczny, np. dlatego, że wymaga pewnego wysiłku, aby uzyskać fragmenty prefiksu, funkcja może zwrócić obietnicę, aby zasygnalizować sukces lub porażkę. Odrzucona obietnica spowoduje błąd strumienia. Wszelkie zgłoszone wyjątki zostaną ponownie zgłoszone przez konstruktor TransformStream().
• transform(chunk, controller): ta metoda jest wywoływana, gdy nowy fragment pierwotnie zapisany po stronie z możliwością zapisu jest gotowy do przekształcenia. Implementacja strumienia zapewnia, że ta funkcja zostanie wywołana tylko po pomyślnym zakończeniu poprzednich przekształceń i nigdy przed zakończeniem działania funkcji start() ani po wywołaniu funkcji flush(). Ta funkcja wykonuje rzeczywistą pracę związaną z przekształcaniem strumienia przekształceń. Może umieszczać wyniki w kolejce za pomocą funkcji controller.enqueue(). Dzięki temu pojedynczy fragment zapisany po stronie z możliwością zapisu może spowodować powstanie zera lub wielu fragmentów po stronie z możliwością odczytu, w zależności od tego, ile razy zostanie wywołana funkcja controller.enqueue(). Jeśli proces przekształcania jest asynchroniczny, ta funkcja może zwrócić obietnicę, aby zasygnalizować powodzenie lub niepowodzenie przekształcenia. Odrzucona obietnica spowoduje błąd zarówno po stronie odczytu, jak i zapisu strumienia przekształcania. Jeśli nie podasz żadnej transform() metody, użyta zostanie transformacja tożsamości, która umieszcza w kolejce bloki bez zmian po stronie zapisu i odczytu.
• flush(controller): ta metoda jest wywoływana po tym, jak wszystkie fragmenty zapisane po stronie zapisu zostaną przekształcone przez pomyślne przejście przez transform(), a strona zapisu ma zostać zamknięta. Zwykle służy to do umieszczania w kolejce fragmentów sufiksu po stronie odczytu, zanim ta strona również zostanie zamknięta. Jeśli proces opróżniania jest asynchroniczny, funkcja może zwrócić obietnicę, aby zasygnalizować powodzenie lub niepowodzenie. Wynik zostanie przekazany do wywołującego funkcji stream.writable.write(). Odrzucona obietnica spowoduje też błąd po stronie odczytu i zapisu strumienia. Wyjątek jest traktowany tak samo jak zwrócenie odrzuconego obiektu Promise.

const transformStream = new TransformStream({
  start(controller) {
    /* … */
  },

  transform(chunk, controller) {
    /* … */
  },

  flush(controller) {
    /* … */
  },
});

Strategie kolejkowania writableStrategy i readableStrategy

Drugi i trzeci opcjonalny parametr konstruktora TransformStream() to opcjonalne strategiewritableStrategy i readableStrategy kolejkowania. Są one zdefiniowane w sposób opisany w sekcjach dotyczących strumieni do odczytu i do zapisu.

Przykładowy kod strumienia przekształceń

Poniższy przykładowy kod pokazuje działanie strumienia przekształcającego.

// Note that `TextEncoderStream` and `TextDecoderStream` exist now.
// This example shows how you would have done it before.
const textEncoderStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

(async () => {
  const readStream = textEncoderStream.readable;
  const writeStream = textEncoderStream.writable;

  const writer = writeStream.getWriter();
  for (const char of 'abc') {
    writer.write(char);
  }
  writer.close();

  const reader = readStream.getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

Przesyłanie strumienia do odczytu przez strumień przekształcający

Metoda pipeThrough() interfejsu ReadableStream umożliwia łączenie strumienia bieżącego z przekształcającym strumieniem lub dowolną inną parą strumieni do zapisu i odczytu. Przekierowanie strumienia zwykle blokuje go na czas przekierowania, uniemożliwiając innym czytnikom jego zablokowanie.

const transformStream = new TransformStream({
  transform(chunk, controller) {
    console.log('[transform]', chunk);
    controller.enqueue(new TextEncoder().encode(chunk));
  },
  flush(controller) {
    console.log('[flush]');
    controller.terminate();
  },
});

const readableStream = new ReadableStream({
  start(controller) {
    // called by constructor
    console.log('[start]');
    controller.enqueue('a');
    controller.enqueue('b');
    controller.enqueue('c');
  },
  pull(controller) {
    // called read when controller's queue is empty
    console.log('[pull]');
    controller.enqueue('d');
    controller.close(); // or controller.error();
  },
  cancel(reason) {
    // called when rs.cancel(reason)
    console.log('[cancel]', reason);
  },
});

(async () => {
  const reader = readableStream.pipeThrough(transformStream).getReader();
  for (let result = await reader.read(); !result.done; result = await reader.read()) {
    console.log('[value]', result.value);
  }
})();

Następny przykładowy kod (trochę naciągany) pokazuje, jak można wdrożyć „krzyczącą” wersję funkcji fetch(), która zamienia wszystkie litery na wielkie, wykorzystując zwróconą obietnicę odpowiedzi jako strumień i zamieniając litery na wielkie partiami. Zaletą tego podejścia jest to, że nie musisz czekać na pobranie całego dokumentu, co może mieć ogromne znaczenie w przypadku dużych plików.

function upperCaseStream() {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

function appendToDOMStream(el) {
  return new WritableStream({
    write(chunk) {
      el.append(chunk);
    }
  });
}

fetch('./lorem-ipsum.txt').then((response) =>
  response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(upperCaseStream())
    .pipeTo(appendToDOMStream(document.body))
);

Prezentacja

Wersja demonstracyjna poniżej pokazuje działanie strumieni do odczytu, zapisu i przekształcania. Zawiera też przykłady pipeThrough() i pipeTo() ciągów znaków potoku oraz pokazuje tee(). Możesz opcjonalnie uruchomić wersję demonstracyjną w osobnym oknie lub wyświetlić kod źródłowy.

Przydatne strumienie dostępne w przeglądarce

W przeglądarce jest wbudowanych kilka przydatnych strumieni. Możesz łatwo utworzyć ReadableStream z obiektu blob. Metoda stream() interfejsu Blob zwraca obiekt ReadableStream, który po odczytaniu zwraca dane zawarte w obiekcie blob. Pamiętaj też, że obiekt File jest szczególnym rodzajem obiektu Blob i może być używany w każdym kontekście, w którym można używać obiektu blob.

const readableStream = new Blob(['hello world'], { type: 'text/plain' }).stream();

Wersje strumieniowe TextDecoder.decode() i TextEncoder.encode() to odpowiednio TextDecoderStream i TextEncoderStream.

const response = await fetch('https://streams.spec.whatwg.org/');
const decodedStream = response.body.pipeThrough(new TextDecoderStream());

Kompresowanie i dekompresowanie plików jest łatwe dzięki strumieniom przekształceń CompressionStream i DecompressionStream. Poniższy przykładowy kod pokazuje, jak pobrać specyfikację Streams, skompresować ją (gzip) bezpośrednio w przeglądarce i zapisać skompresowany plik bezpośrednio na dysku.

const response = await fetch('https://streams.spec.whatwg.org/');
const readableStream = response.body;
const compressedStream = readableStream.pipeThrough(new CompressionStream('gzip'));

const fileHandle = await showSaveFilePicker();
const writableStream = await fileHandle.createWritable();
compressedStream.pipeTo(writableStream);

Interfejs File System Access APIFileSystemWritableFileStream i eksperymentalne fetch() strumienie żądań to przykłady strumieni z możliwością zapisu.

Interfejs Serial API w dużym stopniu korzysta ze strumieni do odczytu i zapisu.

// Prompt user to select any serial port.
const port = await navigator.serial.requestPort();
// Wait for the serial port to open.
await port.open({ baudRate: 9_600 });
const reader = port.readable.getReader();

// Listen to data coming from the serial device.
while (true) {
  const { value, done } = await reader.read();
  if (done) {
    // Allow the serial port to be closed later.
    reader.releaseLock();
    break;
  }
  // value is a Uint8Array.
  console.log(value);
}

// Write to the serial port.
const writer = port.writable.getWriter();
const data = new Uint8Array([104, 101, 108, 108, 111]); // hello
await writer.write(data);
// Allow the serial port to be closed later.
writer.releaseLock();

Na koniec interfejs WebSocketStream API integruje strumienie z interfejsem WebSocket API.

const wss = new WebSocketStream(WSS_URL);
const { readable, writable } = await wss.connection;
const reader = readable.getReader();
const writer = writable.getWriter();

while (true) {
  const { value, done } = await reader.read();
  if (done) {
    break;
  }
  const result = await process(value);
  await writer.write(result);
}

Przydatne materiały

• Specyfikacja strumieni
• Prezentacje
• Wypełnienie interfejsu Streams API
• 2016 r. – rok strumieni danych z sieci
• Iteratory i generatory asynchroniczne
• Wizualizator strumienia

Podziękowania

Ten artykuł został sprawdzony przez Jake’a Archibalda, François Beauforta, Sama Duttona, Mattiasa Buelensa, Surma, Joe Medleya i Adama Rice’a. Posty na blogu Jake'a Archibalda bardzo pomogły mi w zrozumieniu strumieni. Niektóre przykłady kodu zostały zainspirowane eksploracjami użytkownika GitHuba @bellbind, a część tekstu opiera się na dokumentacji MDN Web Docs dotyczącej interfejsu Streams API. Autorzy specyfikacji Streams Standard wykonali świetną robotę.