модуль 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 106
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 101
def stream_for(model, callback = nil, coder: nil, &block)
  stream_from(broadcasting_for(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 76
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

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