Канал Action Cable Streams

Streams позволяют каналам маршрутизировать рассылки подписчикам. Рассылка — это, как обсуждалось ранее, очередь pubsub, где любые данные, помещённые в неё, автоматически отправляются подключённым клиентам. Однако, это чисто онлайн-очередь. Если вы не транслируете рассылку в момент отправки обновления, вы не получите это обновление, даже если подключитесь после его отправки.

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

class CommentsChannel < ApplicationCable::Channel
  def follow(data)
    stream_from "comments_for_#{data['recording_id']}"
  end

  def unfollow
    stop_all_streams
  end
end

Основываясь на приведённом примере, подписчики этого канала получат любые данные, помещённые в, скажем, comments_for_45 рассылку, как только они будут помещены.

Пример рассылки для этого канала выглядит следующим образом:

ActionCable.server.broadcast "comments_for_45", { author: 'DHH', content: 'Rails is just swell' }

Если у вас есть поток, связанный с моделью, то используемая рассылка может быть сгенерирована из модели и канала. Следующий пример подпишется на рассылку, подобную comments:Z2lkOi8vVGVzdEFwcC9Qb3N0LzE.

class CommentsChannel < ApplicationCable::Channel
  def subscribed
    post = Post.find(params[:id])
    stream_for post
  end
end

Вы можете транслировать в этот канал, используя:

CommentsChannel.broadcast_to(@post, @comment)

Если вы не хотите просто передавать рассылку подписчику без фильтрации, вы также можете указать обратный вызов, позволяющий изменить отправляемые данные. Приведённый ниже пример демонстрирует, как это можно использовать для предоставления интроспекции производительности в этом процессе:

class ChatChannel < ApplicationCable::Channel
  def subscribed
    @room = Chat::Room[params[:room_number]]

    stream_for @room, coder: ActiveSupport::JSON do |message|
      if message['originated_at'].present?
        elapsed_time = (Time.now.to_f - message['originated_at']).round(2)

        ActiveSupport::Notifications.instrument :performance, measurement: 'Chat.message_delay', value: elapsed_time, action: :timing
        logger.info "Message took #{elapsed_time}s to arrive"
      end

      transmit message
    end
  end
end

Вы можете остановить трансляцию всех рассылок, вызвав stop_all_streams.