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

Открытые методы экземпляра

stop_all_streams() Показать исходный код

# File actioncable/lib/action_cable/channel/streams.rb, line 104
def stop_all_streams
  streams.each do |broadcasting, callback|
    pubsub.unsubscribe broadcasting, callback
    logger.info "#{self.class.name} stopped streaming from #{broadcasting}"
  end.clear
end

Отписывает все потоки, связанные с этим каналом, от очереди pubsub.

stream_for(model, callback = nil, coder: nil, &block) Показать исходный код

# File actioncable/lib/action_cable/channel/streams.rb, line 99
def stream_for(model, callback = nil, coder: nil, &block)
  stream_from(broadcasting_for([ channel_name, model ]), callback || block, coder: coder)
end

Начать потоковую трансляцию очереди pubsub для model в этом канале. По желанию вы можете передать callback, который будет использоваться вместо стандартной передачи обновлений непосредственно подписчику.

Передайте `coder: ActiveSupport::JSON`, чтобы декодировать сообщения как JSON перед передачей в обратный вызов. По умолчанию `coder: nil`, что не производит декодирования и передаёт исходные сообщения.

stream_from(broadcasting, callback = nil, coder: nil, &block) Показать исходный код

# File actioncable/lib/action_cable/channel/streams.rb, line 74
def stream_from(broadcasting, callback = nil, coder: nil, &block)
  broadcasting = String(broadcasting)

  # Don't send the confirmation until pubsub#subscribe is successful
  defer_subscription_confirmation!

  # Build a stream handler by wrapping the user-provided callback with
  # a decoder or defaulting to a JSON-decoding retransmitter.
  handler = worker_pool_stream_handler(broadcasting, callback || block, coder: coder)
  streams << [ broadcasting, handler ]

  connection.server.event_loop.post do
    pubsub.subscribe(broadcasting, handler, lambda do
      ensure_confirmation_sent
      logger.info "#{self.class.name} is streaming from #{broadcasting}"
    end)
  end
end

Начать потоковую трансляцию из указанной очереди pubsub broadcasting. По желанию вы можете передать callback, который будет использоваться вместо стандартной передачи обновлений непосредственно подписчику. Передайте `coder: ActiveSupport::JSON`, чтобы декодировать сообщения как JSON перед передачей в обратный вызов. По умолчанию `coder: nil`, что не производит декодирования и передаёт исходные сообщения.