123456789_123456789_123456789_123456789_123456789_

Class: ActionCable::SubscriptionAdapter::Redis::Listener

Relationships & Source Files
Namespace Children
Classes:
SubscribedClient
Super Chains via Extension / Inclusion / Inheritance
Class Chain:
self, ::ActionCable::SubscriptionAdapter::SubscriberMap
Instance Chain:
self, ::ActionCable::SubscriptionAdapter::SubscriberMap
Inherits: ActionCable::SubscriptionAdapter::SubscriberMap
Defined in: actioncable/lib/action_cable/subscription_adapter/redis.rb

Constant Summary

Class Method Summary

::ActionCable::SubscriptionAdapter::SubscriberMap - Inherited

.new

Instance Attribute Summary

Instance Method Summary

::ActionCable::SubscriptionAdapter::SubscriberMap - Inherited

#add_channel, #add_subscriber, #broadcast, #invoke_callback, #remove_channel, #remove_subscriber

Constructor Details

.new(adapter, config_options, event_loop) ⇒ Listener

[ GitHub ] 

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 68



def initialize(adapter, config_options, event_loop)
  super()

  @adapter = adapter
  @event_loop = event_loop

  @subscribe_callbacks = Hash.new { |h, k| h[k] = [] }
  @subscription_lock = Mutex.new

  @reconnect_attempt = 0
  # Use the same config as used by Redis conn
  @reconnect_attempts = config_options.fetch(:reconnect_attempts, 1)
  @reconnect_attempts = Array.new(@reconnect_attempts, 0) if @reconnect_attempts.is_a?(Integer)

  @subscribed_client = nil

  @when_connected = []

  @thread = nil
end


  







Instance Attribute Details



  

    #retry_connecting?  ⇒ Boolean  (readonly, private)
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 185



def retry_connecting?
  @reconnect_attempt += 1

  return false if @reconnect_attempt > @reconnect_attempts.size

  sleep_t = @reconnect_attempts[@reconnect_attempt - 1]

  sleep(sleep_t) if sleep_t > 0

  true
end


  







Instance Method Details



  

    #add_channel(channel, on_success)  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 141



def add_channel(channel, on_success)
  @subscription_lock.synchronize do
    ensure_listener_running
    @subscribe_callbacks[channel] << on_success
    when_connected { @subscribed_client.subscribe(channel) }
  end
end


  








  

    #ensure_listener_running   (private)
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 160



def ensure_listener_running
  @thread ||= Thread.new do
    Thread.current.abort_on_exception = true

    begin
      conn = @adapter.redis_connection_for_subscriptions
      listen conn
    rescue ConnectionError
      reset
      if retry_connecting?
        when_connected { resubscribe }
        retry
      end
    end
  end
end


  








  

    #extract_subscribed_client(conn)   (private)
  



  

    

See additional method definition at line 243.


  



  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 249



def extract_subscribed_client(conn)
  SubscribedClient.new(conn._client)
end


  








  

    #invoke_callback  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 155



def invoke_callback(*)
  @event_loop.post { super }
end


  








  

    #listen(conn)  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 89



def listen(conn)
  conn.without_reconnect do
    original_client = extract_subscribed_client(conn)

    conn.subscribe("_action_cable_internal") do |on|
      on.subscribe do |chan, count|
        @subscription_lock.synchronize do
          if count == 1
            @reconnect_attempt = 0
            @subscribed_client = original_client

            until @when_connected.empty?
              @when_connected.shift.call
            end
          end

          if callbacks = @subscribe_callbacks[chan]
            next_callback = callbacks.shift
            @event_loop.post(&next_callback) if next_callback
            @subscribe_callbacks.delete(chan) if callbacks.empty?
          end
        end
      end

      on.message do |chan, message|
        broadcast(chan, message)
      end

      on.unsubscribe do |chan, count|
        if count == 0
          @subscription_lock.synchronize do
            @subscribed_client = nil
          end
        end
      end
    end
  end
end


  








  

    #remove_channel(channel)  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 149



def remove_channel(channel)
  @subscription_lock.synchronize do
    when_connected { @subscribed_client.unsubscribe(channel) }
  end
end


  








  

    #reset   (private)
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 204



def reset
  @subscription_lock.synchronize do
    @subscribed_client = nil
    @subscribe_callbacks.clear
    @when_connected.clear
  end
end


  








  

    #resubscribe   (private)
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 197



def resubscribe
  channels = @sync.synchronize do
    @subscribers.keys
  end
  @subscribed_client.subscribe(*channels) unless channels.empty?
end


  








  

    #shutdown  
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 128



def shutdown
  @subscription_lock.synchronize do
    return if @thread.nil?

    when_connected do
      @subscribed_client.unsubscribe
      @subscribed_client = nil
    end
  end

  Thread.pass while @thread.alive?
end


  








  

    #when_connected(&block)   (private)
  

  [ GitHub ]


  

  


# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 177



def when_connected(&block)
  if @subscribed_client
    block.call
  else
    @when_connected << block
  end
end