Action Cable Streams

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

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

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.