WebSocketStream: интеграция потоков с WebSocket API

# Общая информация

# WebSocket API

WebSocket API предоставляет интерфейс JavaScript для протокола WebSocket, что позволяет открывать сеанс двусторонней интерактивной передачи данных между браузером пользователя и сервером. Используя этот API, можно отправлять сообщения на сервер и получать ответы на основе событий, не опрашивая сервер.

# Streams API

Streams API позволяет JavaScript программно получать доступ к потокам фрагментов данных, полученных по сети, и обрабатывать их. Важным понятием в контексте потоков является обратная реакция — процесс, посредством которого отдельный поток или цепочка перенаправления регулирует скорость чтения или записи. Когда исходный поток или поток далее в цепочке перенаправления еще занят и не готов принимать следующие фрагменты, он отправляет обратно по цепочке сигнал о необходимости замедлить доставку данных.

# Проблема с текущим WebSocket API

# Невозможность применять обратную реакцию к полученным сообщениям

В текущем WebSocket API реакция на сообщение происходит в WebSocket.onmessage — обработчике EventHandler, который вызывается при получении сообщения от сервера.

Предположим, у нас есть приложение, которое с каждым полученным сообщением должно выполнять ресурсоемкую обработку данных. Соответствующий код наверняка будет похож на приведенный ниже. В нем мы ожидаем (await) результат вызова process(), поэтому всё должно работать, верно?

// Ресурсоемкая обработка данных.
const process = async (data) => {
  return new Promise((resolve) => {
    window.setTimeout(() => {
      console.log('Сообщение WebSocket обработано:', data);
      return resolve('done');
    }, 1000);
  });
};

webSocket.onmessage = async (event) => {
  const data = event.data;
  // Ждем результат этапа обработки в обработчике сообщений.
  await process(data);
};

Неверно! Проблема с текущим WebSocket API в том, что мы не можем использовать обратную реакцию. Если сообщения приходят быстрее, чем их может обработать метод process(), процесс отрисовки либо забьет память, буферизируя эти сообщения, либо перестанет отвечать из-за 100 % загрузки ЦП (или произойдет и то, и другое).

# Применять обратную реакцию к отправляемым сообщениям — неудобно

Применять обратную реакцию к отправляемым сообщениям можно, но для этого нужно запрашивать свойство WebSocket.bufferedAmount, что неэффективно и неудобно. Это доступное только для чтения свойство возвращает количество байтов данных, помещенных в очередь посредством вызовов WebSocket.send(), но еще не переданных в сеть. Значение сбрасывается на ноль после отправки всех данных в очереди, но если продолжить вызывать WebSocket.send(), оно будет повышаться.

# Что представляет собой WebSocketStream API

WebSocketStream API решает проблему отсутствия надлежащей обратной реакции путем интеграции потоков с WebSocket API. То есть, обратную реакцию можно применять «бесплатно» — без дополнительных ресурсозатрат.

# Предлагаемые варианты использования для WebSocketStream API

Примеры сайтов, которые могут использовать этот API:

• Приложения WebSocket с высокой сетевой нагрузкой, которые должны оставаться интерактивными, в частности — решения для трансляции видео и демонстрации экрана.
• Средства для видеозахвата и другие приложения, генерирующие в браузере большие объемы данных, который необходимо загрузить на сервер. Используя обратную реакцию, клиент может остановить производство данных и не накапливать их в памяти.

# Текущее состояние

# Как использовать WebSocketStream API

# Вводный пример

WebSocketStream API работает на промисах, и поэтому естественным образом встраивается в современный мир JavaScript. Для начала мы создаем новый объект WebSocketStream передаем ему URL-адрес сервера WebSocket. Затем ждем , пока установится подключение (connection). В результате мы получаем ReadableStream и (или) WritableStream.

Вызывая метод ReadableStream.getReader(), мы в итоге получаем ReadableStreamDefaultReader, с помощью которого можно читать (read()) данные из потока до его завершения, то есть пока поток не вернет объект в форме {value: undefined, done: true}.

Аналогичным образом, вызывая метод WritableStream.getWriter(), мы в итоге получаем WritableStreamDefaultWriter, с помощью которого можно записывать (write()) данные.

  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);
  }

# Обратная реакция

А что насчет обещанной функции обратной реакции? Как я писал выше, она дается «бесплатно» — ничего дополнительно делать не нужно: если для process() требуется больше времени, следующее сообщение будет потреблено, только когда конвейер будет готов. Соответственно, и действие WritableStreamDefaultWriter.write() будет выполняться только в том случае, когда это допустимо.

# Более сложные примеры

Второй аргумент в WebSocketStream — это набор параметров, число которых может быть увеличено в будущем. Пока что есть только параметр protocols, который ведет себя так же, как второй аргумент в конструкторе WebSocket:

const chatWSS = new WebSocketStream(CHAT_URL, {protocols: ['chat', 'chatv2']});
const {protocol} = await chatWSS.connection;

Выбранный протокол (protocol) и возможные расширения (extensions) входят в словарь, доступный по промису WebSocketStream.connection. В этом промисе содержится вся информация о действующем подключении (если связь не установлена, эти данные неактуальны).

const {readable, writable, protocol, extensions} = await chatWSS.connection;

# Информация о закрытом подключении WebSocketStream

Информацию, которая была доступна в событиях WebSocket.onclose и WebSocket.onerror в WebSocket API, теперь можно получить через промис WebSocketStream.closed. Если подключение закрыто ненадлежащим образом, промис отклоняется. В противном случае он разрешается с кодом и причиной, полученными от сервера.

Все возможные коды состояния и их значение объяснены в списке для CloseEvent.

const {code, reason} = await chatWSS.closed;

# Закрытие подключения WebSocketStream

Закрыть WebSocketStream можно с помощью AbortController: достаточно передать AbortSignal в конструктор WebSocketStream.

const controller = new AbortController();
const wss = new WebSocketStream(URL, {signal: controller.signal});
setTimeout(() => controller.abort(), 1000);

Также можно использовать метод WebSocketStream.close(), но его основное предназначение — указание кода и причины, которые отправляются на сервер.

wss.close({code: 4000, reason: 'Игра закончена'});

# Прогрессивное улучшение и совместимость

Пока что WebSocketStream API реализован только в браузере Chrome. Для совместимости с классическим WebSocket API применение обратной реакции к полученным сообщениям невозможно. Применять обратную реакцию к отправляемым сообщениям можно, но для этого нужно запрашивать свойство WebSocket.bufferedAmount, что неэффективно и неудобно.

# Обнаружение функции

Как проверить, поддерживается ли WebSocketStream API:

if ('WebSocketStream' in window) {
  // `WebSocketStream` поддерживается!
}

# Демонстрация

Увидеть работу WebSocketStream API (если поддержка в браузере реализована) можно во встроенном iframe здесь или на странице Glitch.

# Отзывы

Команде Chrome хотелось бы услышать ваши отзывы о работе с WebSocketStream API.

# Расскажите, что вы думаете о структуре API

Что-то в API не работает должным образом? Или, может, отсутствуют методы или свойства, необходимые для реализации вашей идеи? Есть вопрос или комментарий по модели безопасности? Отправьте заявку о проблеме со спецификацией (Spec issue) в GitHub-репозиторий или прокомментируйте существующую.

# Сообщите о проблеме с реализацией

Нашли ошибку в реализации функции в браузере Chrome? Реализация отличается от спецификации? Сообщите об ошибке на странице new.crbug.com. Опишите проблему как можно подробнее, дайте простые инструкции по ее воспроизведению и в поле Components укажите Blink>Network>WebSockets. Для демонстрации этапов воспроизведения ошибки удобно использовать Glitch.

# Окажите поддержку API

Планируете использовать WebSocketStream API? Ваш публичный интерес помогает команде Chrome определять приоритет функций и показывает важность их поддержки разработчикам других браузеров.

Упомяните в твите @ChromiumDev, поставьте хештег #WebSocketStream и расскажите, как вы используете эту функцию.

# Полезные ссылки

• Публичное пояснение.
• Демонстрация WebSocketStream API. | Исходный код демопримера WebSocketStream API.
• Отслеживание ошибок.
• Запись на ChromeStatus.com.
• Компонент Blink: Blink>Network>WebSockets.

# Благодарности

Авторы реализации WebSocketStream API — Адам Райс и Ютака Хирано. Автор главного изображения — Даан Муйдж, платформа Unsplash.