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

Предотвратите перегрузку вашего приложения сообщениями WebSocket или заваливание WebSocket-сервера сообщениями, используя механизм обратного давления.

Thomas Steiner

Фон

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

API потоков

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

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

Применение обратного давления к полученным сообщениям невозможно.

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

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

// A heavy data crunching operation.
const process = async (data) => {
  return new Promise((resolve) => {
    window.setTimeout(() => {
      console.log('WebSocket message processed:', data);
      return resolve('done');
    }, 1000);
  });
};

webSocket.onmessage = async (event) => {
  const data = event.data;
  // Await the result of the processing step in the message handler.
  await process(data);
};

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

Применение обратного давления к отправленным сообщениям неэргономично.

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

Что такое API WebSocketStream?

API WebSocketStream решает проблему отсутствия или неэргономичного обратного давления за счет интеграции потоков с API WebSocket. Это означает, что обратное давление может применяться «бесплатно», без каких-либо дополнительных затрат.

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

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

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

Текущий статус

Шаг	Статус
1. Создайте пояснительное видео.	Полный
2. Создайте первоначальный черновик спецификации.	В ходе выполнения
3. Сбор отзывов и итеративная доработка дизайна.	В ходе выполнения
4. Испытание происхождения	Полный
5. Запуск	Не началось

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

API WebSocketStream основан на промисах, что делает работу с ним естественной в современном мире JavaScript. Вы начинаете с создания нового WebSocketStream и передачи ему URL-адреса WebSocket-сервера. Затем вы ждете opened соединения, в результате чего получаете объект 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.opened;
  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.opened;

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

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

Примечание: В более ранней версии API метод WebSocketStream.opened назывался WebSocketStream.connection .

Информация о закрытом соединении WebSocketStream

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

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

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

Закрытие соединения WebSocketStream

Объект WebSocketStream можно закрыть с помощью AbortController . Поэтому передайте AbortSignal в конструктор WebSocketStream . AbortController.abort() работает только до установления соединения, а не после.

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

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

wss.close({closeCode: 4000, reason: 'Game over'});

Постепенное совершенствование и обеспечение совместимости

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

Обнаружение признаков

Чтобы проверить, поддерживается ли API WebSocketStream, используйте:

if ('WebSocketStream' in window) {
  // `WebSocketStream` is supported!
}

Демо

В поддерживающих браузерах вы можете увидеть API WebSocketStream в действии или посмотреть встроенный iframe.

Обратная связь

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

Расскажите о проектировании API.

Есть ли что-то в API, что работает не так, как вы ожидали? Или отсутствуют методы или свойства, необходимые для реализации вашей идеи? Есть вопрос или комментарий по модели безопасности? Создайте заявку в соответствующем репозитории GitHub или добавьте свои мысли в существующую заявку.

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

Вы обнаружили ошибку в реализации Chrome? Или реализация отличается от спецификации? Сообщите об ошибке на new.crbug.com . Обязательно укажите как можно больше подробностей, простые инструкции по воспроизведению и введите Blink>Network>WebSockets в поле «Компоненты» .

Показать поддержку API

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

Отправьте твит @ChromiumDev , используя хэштег #WebSocketStream , и расскажите, где и как вы его используете.

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

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

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

API WebSocketStream был реализован Адамом Райсом и Ютакой Хирано .