Использование потоков записи

В качестве разработчика JavaScript, программно записывать данные в поток очень полезно! Эта статья объясняет функциональность потоков записи API потоков.

В нашем репозитории dom-examples/streams вы найдете пример простой записи (посмотреть онлайн). Он принимает заданное сообщение и записывает его в поток записи, отображая каждый фрагмент на пользовательском интерфейсе по мере записи в поток, а также отображает все сообщение на пользовательском интерфейсе после завершения записи.

Давайте рассмотрим, как работает функциональность потоков записи в нашем демонстрационном примере.

Чтобы создать поток записи, мы используем конструктор WritableStream(); синтаксис на первый взгляд сложный, но на самом деле не так уж и плох.

Схема синтаксиса выглядит так:

const stream = new WritableStream(
  {
    start(controller) {},
    write(chunk, controller) {},
    close(controller) {},
    abort(reason) {},
  },
  {
    highWaterMark: 3,
    size: () => 1,
  },
);

Конструктор принимает два объекта в качестве параметров. Первый объект обязателен и создает в JavaScript модель базового приемника, в который записываются данные. Второй объект необязателен и позволяет указать стратегию очереди для вашего потока, которая принимает форму экземпляра ByteLengthQueuingStrategy или CountQueuingStrategy.

Первый объект может содержать до четырёх членов, все из которых необязательны:

1. start(controller) — метод, вызываемый один раз сразу после создания потока WritableStream. В этом методе вы должны включить код, который настраивает функциональность потока, например, получение доступа к базовому приемнику.
2. write(chunk,controller) — метод, вызываемый многократно каждый раз, когда новый фрагмент готов к записи в базовый приемник (указанный в параметре chunk).
3. close(controller) — метод, вызываемый, если приложение сигнализирует о завершении записи фрагментов в поток. Он должен выполнить все необходимые действия для завершения записи в базовый приемник и освобождения доступа к нему.
4. abort(reason) — метод, который будет вызван, если приложение сигнализирует о том, что хочет внезапно закрыть поток и перевести его в состояние ошибки.

Вызов конструктора в нашем примере выглядит так:

const decoder = new TextDecoder("utf-8");
const queuingStrategy = new CountQueuingStrategy({ highWaterMark: 1 });
let result = "";
const writableStream = new WritableStream(
  {
    // Implement the sink
    write(chunk) {
      return new Promise((resolve, reject) => {
        const buffer = new ArrayBuffer(1);
        const view = new Uint8Array(buffer);
        view[0] = chunk;
        const decoded = decoder.decode(view, { stream: true });
        const listItem = document.createElement("li");
        listItem.textContent = `Chunk decoded: ${decoded}`;
        list.appendChild(listItem);
        result += decoded;
        resolve();
      });
    },
    close() {
      const listItem = document.createElement("li");
      listItem.textContent = `[MESSAGE RECEIVED] ${result}`;
      list.appendChild(listItem);
    },
    abort(err) {
      console.error("Sink error:", err);
    },
  },
  queuingStrategy,
);

• Метод write() содержит промис с кодом, который декодирует каждый записанный фрагмент в формат, который можно записать на пользовательском интерфейсе. Он вызывается, когда каждый фрагмент фактически записывается (см. следующий раздел).
• Метод close() вызывается автоматически при завершении записи — он выводит весь декодированный результат на пользовательский интерфейс в виде одной строки.
• Метод abort() выводит сообщение об ошибке в консоль, если поток прерван.

Чтобы фактически записать содержимое в поток, мы вызываем функцию sendMessage(), передавая ей сообщение для записи и поток для записи:

sendMessage("Hello, world.", writableStream);

Определение sendMessage() выглядит следующим образом:

function sendMessage(message, writableStream) {
  // defaultWriter is of type WritableStreamDefaultWriter
  const defaultWriter = writableStream.getWriter();
  const encoder = new TextEncoder();
  const encoded = encoder.encode(message, { stream: true });
  encoded.forEach((chunk) => {
    defaultWriter.ready
      .then(() => defaultWriter.write(chunk))
      .then(() => console.log("Chunk written to sink."))
      .catch((err) => console.error("Chunk error:", err));
  });
  // Call ready again to ensure that all chunks are written
  //   before closing the writer.
  defaultWriter.ready
    .then(() => defaultWriter.close())
    .then(() => console.log("All chunks written"))
    .catch((err) => console.error("Stream error:", err));
}

Итак, здесь мы создаём записывающее устройство для записи фрагментов в поток с помощью WritableStream.getWriter(). Это создаёт экземпляр WritableStreamDefaultWriter.

Мы также создаём новый экземпляр TextEncoder, используя соответствующий конструктор для кодирования сообщения в фрагменты, которые будут помещены в поток.

После кодирования фрагментов мы вызываем forEach() на получившемся массиве. Внутри этого блока мы используем WritableStreamDefaultWriter.ready для проверки готовности записывающего устройства к записи ещё одного фрагмента. ready возвращает промис, который выполняется, когда это так, внутри которого мы вызываем WritableStreamDefaultWriter.write() для фактической записи фрагмента в поток. Это также запускает метод write() , указанный внутри конструктора WritableStream(), как обсуждалось выше.

После записи всех фрагментов мы снова выполняем проверку ready, чтобы проверить, что последний фрагмент был записан и вся работа завершена. Когда эта проверка ready выполняется, мы вызываем WritableStreamDefaultWriter.close() для закрытия потока. Это также запускает метод close() , указанный внутри конструктора WritableStream(), как обсуждалось выше.

Как вы заметили, изучая схему синтаксиса WritableStream(), методы start(), write(), и close() могут необязательно принимать параметр controller. Он содержит экземпляр интерфейса WritableStreamDefaultController, который может использоваться разработчиком для дальнейшего управления потоком по мере необходимости.

В настоящее время у него доступен только один метод — WritableStreamDefaultController.error(), который, при вызове, приводит к ошибке будущих взаимодействий с потоком. Это полезно, когда другая часть приложения работает неправильно, и вы хотите распространить ошибку на поток, чтобы вся система завершилась корректно, а не рисковать записью мусора в поток (или чем-то подобным).

Как упоминалось выше, мы вызываем метод close() при завершении записи, что запускает метод close() , указанный внутри конструктора WritableStream().

Мы также можем прервать поток, вызвав WritableStreamDefaultWriter.abort().

Разница заключается в том, что при вызове close любые ранее поставленные в очередь фрагменты записываются и завершаются, прежде чем поток будет закрыт.

При вызове abort любые ранее поставленные в очередь фрагменты просто отбрасываются сразу, а затем поток переводится в состояние ошибки. Это также вызывает вызов любого метода abort() , указанного в конструкторе WritableStream().