Реальное время: потоки в браузер

01Как рассылка доходит до страницы

Страница подписывается на один или несколько именованных потоков; любая часть приложения рассылает сообщение в поток по его имени; каждая подписанная страница получает его примерно за десятую долю секунды.

• У потоков есть имена. Имя — любой короткий текст ("alerts") или запись, чьё имя — её таблица и id ("articles:12", Cybertrain::Stream.name_for(@article)).
• Рассылка — это строка таблицы. Cybertrain::Stream.broadcast(name, payload) вставляет строку в cybertrain_messages, таблицу, которую фреймворк сам создаёт в базе данных приложения. Внутри DB.transaction сообщение уходит, только если транзакция зафиксирована. Раз это строка, рассылка работает из любого процесса с той же базой: из запроса, из задачи в процессе worker, из колбэка фида, из Main.command.
• Каждый веб-процесс опрашивает таблицу. Один поток на процесс server (опросчик потоков) каждые stream_poll_ms (100 мс) читает новые строки и передаёт каждую браузерам, которые держит этот процесс. Раз в минуту он удаляет строки старше stream_retention (60 секунд).
• Браузер подключается к одной точке входа, /cybertrain/stream, через Server-Sent Events (EventSource) или WebSocket, с подписанным токеном, который называет потоки, доступные ему для чтения (Подписанные имена потоков).

Запускать ничего не нужно: ./<app> server запускает опросчик и обслуживает точку входа. config.stream_path = "" выключает всю функцию.

02Turbo Streams

Turbo обновляет страницу по элементам <turbo-stream action="append" target="comments">, как это делают приложения на Rails 8. Подключите Turbo в макете (конвейера ресурсов нет: CDN или копия в public/) и подпишите страницу с помощью turbo_stream_from:

<head>
  <%= csrf_meta_tags %>
  <script type="module" src="https://cdn.jsdelivr.net/npm/@hotwired/turbo@8/dist/turbo.es2017-esm.js"></script>
</head>

<%= turbo_stream_from @article %>
<div id="comments">
  <% @article.comments.each do |comment| %>
    <%= render "comments/comment", comment: comment %>
  <% end %>
</div>

turbo_stream_from принимает строки и записи, сколько угодно, и выводит один элемент <turbo-stream-source>, чей URL несёт один токен для всех них. Turbo открывает его как EventSource; turbo_stream_from @article, ws: true вместо этого даёт URL ws:// (wss:// за HTTPS), который Turbo открывает как WebSocket. В обоих случаях странице не нужен ваш JavaScript. Потерянный WebSocket (обрыв сети, перезапуск сервера) Turbo заново не открывает: такая страница не получает обновлений, пока её не перезагрузят, тогда как EventSource переподключается сам и повторяет пропущенное. Оставьте вариант по умолчанию, если WebSocket не нужен, или переподключайтесь из своего JavaScript (&after=, ниже).

Затем разошлите сообщение из действия, которое что-то изменило:

def create
  @comment = Comment.new(comment_params)
  @comment.article_id = @article.id
  if @comment.save
    Cybertrain::Stream.broadcast_append_to(@article, target: "comments",
                                           partial: "comments/comment", locals: { comment: @comment })
  end
  redirect_to article_path(@article), status: :see_other
end

broadcast_append_to(stream, target:, ...) отрисовывает содержимое, оборачивает его в элемент <turbo-stream> и рассылает. Содержимое — одно из: partial: с locals: (отрисовывается Cybertrain::Views.render_partial), component: (компонент представления) или html: (экранируется, если это не SafeString). Цель — DOM id или запись, чей DOM id — dom_id(record) ("comment_7"). Остальные работают так же: broadcast_replace_to, broadcast_update_to, broadcast_prepend_to, broadcast_before_to, broadcast_after_to, broadcast_remove_to(stream, target:) и broadcast_refresh_to(stream) (Turbo 8 перезагружает страницу с морфингом).

Turbo также превращает ссылки и формы в запросы fetch (Turbo Drive). Scaffold уже отвечает так, как ожидает Turbo: редирект с 303 после успешной формы и 422 с формой снова после неудачной.

Ответ на форму через Turbo Stream

Форма, которую отправляет Turbo, принимает text/vnd.turbo-stream.html. Чтобы ответить потоком вместо редиректа, отрисуйте элемент сами; их строит Cybertrain::Turbo (replace, update, append, prepend, before, after, remove, refresh):

def destroy
  @comment = Comment.where(article_id: @article.id).find(params[:id])
  @comment.destroy
  if request.header("accept").to_s.include?("text/vnd.turbo-stream.html")
    render html: Cybertrain::Turbo.remove(@comment), content_type: "text/vnd.turbo-stream.html"
  else
    redirect_to article_path(@article), status: :see_other
  end
end

Рассылка из задачи, периодической задачи или фида

Рассылке не нужен запрос, поэтому задача может отрисовать и отправить её:

Cybertrain::Job.handle("price_refresh") do |payload|
  item = Item.find(Cybertrain::Cast.int(payload))
  item.refresh_price!
  Cybertrain::Stream.broadcast_replace_to("prices", target: item,
                                          partial: "items/price", locals: { item: item })
end

Партиал отрисовывает Cybertrain::Views.render_partial, который работает без запроса: хелперы путей (item_path(item)) работают, а хелперы, которым нужен запрос (csrf_token, params, flash, request_path, *_url), возбуждают Cybertrain::Views::OutsideRequest (см. Отрисовка вне запроса). Задача выполняется и в процессе worker, и на сервере, а строка, которую она пишет, доходит до страниц всех серверов.

03Server-Sent Events из JavaScript

Для страницы, которой управляет не Turbo (график, бегущая строка), запрашивайте потоки в JSON. stream_source_path даёт путь точки входа с токеном и mode=json, а signed_stream_name — только токен:

<div id="price" data-src="<%= stream_source_path "prices", "alerts" %>"></div>
<script type="module">
  const el = document.getElementById("price");
  const source = new EventSource(el.dataset.src);
  source.addEventListener("prices", (event) => {
    const message = JSON.parse(event.data);   // {"stream":"prices","id":42,"data":"..."}
    el.textContent = JSON.parse(message.data).price;
  });
  source.addEventListener("alerts", (event) => console.log(JSON.parse(event.data).data));
</script>

В mode=json каждое сообщение — событие с именем своего потока, данные которого — {"stream": ..., "id": ..., "data": ...}, с полезной нагрузкой в data (рассылайте JSON как String: Stream.broadcast("prices", JSON.generate({ "price" => 12.5 }))). В режиме по умолчанию mode=raw нагрузка уходит как есть, обычным событием message, — это то, что читает Turbo.

Когда соединение рвётся, браузер переподключается сам и отправляет id последнего увиденного события (Last-Event-ID). Тогда точка входа сначала повторно отправляет все сообщения этих потоков с бо́льшим id, которые ещё есть в таблице (в пределах stream_retention), так что ничего из промежутка не теряется. Странице, отсутствовавшей дольше, стоит перезагрузиться. До прихода первого события браузеру нечего отправить в качестве id, поэтому соединение, оборвавшееся в этом окне, ничего не получает повторно: то, что страница отрисовала при загрузке, остаётся её состоянием до следующей рассылки. После 15 секунд тишины уходит строка комментария (: ping), чтобы прокси держали соединение открытым.

04WebSocket из JavaScript

Тот же URL отвечает и на WebSocket. Страница строит из него URL ws:// или wss://. Передавайте путь и токены через атрибуты data-, как здесь: <%= %> внутри элемента <script> экранируется как HTML, а браузер там его не декодирует (на сервер пришло бы &).

<div id="ticker" data-src="<%= stream_source_path "prices" %>" data-alerts="<%= signed_stream_name "alerts" %>"></div>
<script type="module">
  const el = document.getElementById("ticker");
  const scheme = location.protocol === "https:" ? "wss://" : "ws://";
  const socket = new WebSocket(scheme + location.host + el.dataset.src);
  socket.addEventListener("message", (event) => {
    const message = JSON.parse(event.data);
    if (message.stream) console.log(message.stream, message.id, message.data);
  });
  socket.addEventListener("open", () => {
    socket.send(JSON.stringify({ subscribe: el.dataset.alerts }));
  });
</script>

В mode=json каждое сообщение — один текстовый фрейм с тем же конвертом {"stream", "id", "data"}, и страница может менять свои потоки во время соединения, отправляя текстовый фрейм JSON:

Соединение в mode=json может также открыться без токена и подписаться сообщением; соединение, которое через 15 секунд всё ещё не читает ни одного потока, закрывается (код 1008). У WebSocket нет Last-Event-ID: чтобы наверстать после переподключения, добавьте к URL &after=<the last id you saw>. Сервер отправляет ping после каждых 15 секунд тишины и отвечает на ping; двоичные фреймы от страницы игнорируются. Коды закрытия: 1000, когда сервер останавливается, 1008 для плохого токена, 1009 для фрейма больше stream_max_message, 1013, когда процесс заполнен или страница слишком отстала.

05Потоковая передача из действия

Действие может держать открытым собственный ответ, как это делает ActionController::Live в Rails: stream выполняет блок после действия, в потоке соединения, и передаёт ему сокет как Cybertrain::Sink.

def progress
  stream do |out|
    10.times do |i|
      out.write(Cybertrain::SSE.event(JSON.generate({ "done" => i * 10 }), event: "progress", id: i.to_s))
      break if out.wait_closed(1)   # true, когда клиент ушёл
    end
  end
end

stream(content_type: "text/event-stream", status: :ok) отвечает с Transfer-Encoding: chunked и закрывает соединение, когда блок возвращается. SSE.event(data, event: "", id: "", retry_ms: 0) форматирует одно событие; out.closed? становится true, как только клиент ушёл, и последующие записи ничего не делают. Блок не должен трогать сессию, flash или заголовки: они ушли вместе с началом ответа.

06Подписанные имена потоков

Страница может читать поток только с токеном, который его называет. Токен — это имена потоков и подпись HMAC-SHA256, сделанная ключом, выведенным из secret_key_base только для потоков (как turbo/signed_stream_verifier_key в Rails), поэтому токен никогда не подходит как cookie сессии; точка входа проверяет подпись и не читает ни сессию, ни базу данных. Это модель Turbo: тот, кто мог отрисовать страницу, мог читать её потоки. Токены не истекают; смена secret_key_base делает недействительными все, и страница, открытая через такой деплой, перестаёт получать обновления (EventSource считает 403 окончательным), пока её не перезагрузят.

• Поток для одного пользователя — это поток, названный по записи этого пользователя, turbo_stream_from @user ("users:7"), отрисованный только на страницах этого пользователя. Любой, кто заполучит токен, может читать этот поток, пока секрет не меняется.
• Токены не истекают. Смена CYBERTRAIN_SECRET_KEY_BASE делает недействительными все токены (и все сессии).
• Токен передаётся в URL, поэтому журнал доступа обратного прокси его записывает. Собственный журнал запросов фреймворка пишет его как [FILTERED]; выключите журналирование прокси для /cybertrain/stream или держите эти журналы такими же закрытыми, как потоки.
• Плохой или отсутствующий токен — это 403 для SSE и закрытие с 1008 сразу после рукопожатия для WebSocket.
• Имя — от 1 до 200 байт без управляющих символов и запятых; токен несёт не больше 16 имён, соединение — не больше 64.

07Ограничения и настройки

Задаются в config/app.rb (c.stream_poll_ms = 250); при значении вне диапазона запуск останавливается с сообщением.

Каждый открытый поток занимает одно соединение и его зелёный поток, пока страница открыта. За nginx заголовок ответа X-Accel-Buffering: no выключает буферизацию; для WebSocket прокси должен пропускать Upgrade и Connection (Caddy и kamal-proxy пропускают).

08Какой процесс что делает

• ./<app> server всегда запускает опросчик и точку входа, в том числе с CYBERTRAIN_WORKER=0: он доставляет сообщения браузерам, которые держит. Пока опросчик не работает, точка входа отвечает 503 (stream poller is not running); serve запускает его до открытия порта, так что обслуживающий процесс этого никогда не показывает.
• ./<app> worker и ./<app> feeds не обслуживают HTTP и не запускают опросчик. Их рассылки — это строки, которые доставляют серверы (Роли процессов).
• Несколько серверов на одной базе каждый доставляют каждое сообщение своим браузерам, в одном и том же порядке и с одними и теми же id.

Тестирование потока страницы: Тестирование потоков и компонентов. Устройство: docs/design.md D36 и D37.