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

Не допускайте переполнения вашего приложения сообщениями WebSocket или переполнения сервера WebSocket сообщениями, применяя обратное давление.

Thomas Steiner

Фон

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

API потоков

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

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

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

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

Предположим, у вас есть приложение, которому необходимо выполнять тяжелые операции по обработке данных при получении нового сообщения. Вероятно, вы настроите поток, аналогичный коду ниже, и поскольку вы 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 с высокой пропускной способностью, которым необходимо сохранять интерактивность, в частности видео и совместное использование экрана.
• Аналогично, видеозахват и другие приложения, которые генерируют много данных в браузере, которые необходимо загрузить на сервер. С помощью обратного давления клиент может прекратить производить данные, а не накапливать их в памяти.

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

Как использовать 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;

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

Информация, которая была доступна из событий WebSocket.onclose и WebSocket.onerror в API WebSocket, теперь доступна через обещание 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: 'Game over'});

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

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

Обнаружение особенностей

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

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

Демо

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

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

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

Расскажите нам о дизайне API

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

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

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

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

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

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

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

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

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

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