модуль ActionCable::Channel::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.