модуль 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([ 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 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 , что не производит декодирования, передаёт сообщения в сыром виде.