Потоки канала Action Cable

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.