class ActionCable::SubscriptionAdapter::Redis::Listener
Constants
- ConnectionError
Public Class Methods
new(adapter, config_options, event_loop)
click to toggle source
Calls superclass method
# File 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
Public Instance Methods
add_channel(channel, on_success)
click to toggle source
# File 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
invoke_callback(*)
click to toggle source
Calls superclass method
# File lib/action_cable/subscription_adapter/redis.rb, line 155 def invoke_callback(*) @event_loop.post { super } end
listen(conn)
click to toggle source
# File 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)
click to toggle source
# File 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
shutdown()
click to toggle source
# File 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
Private Instance Methods
ensure_listener_running()
click to toggle source
# File 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)
click to toggle source
# File lib/action_cable/subscription_adapter/redis.rb, line 243 def extract_subscribed_client(conn) SubscribedClient.new(conn._client) end
reset()
click to toggle source
# File 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()
click to toggle source
# File 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
retry_connecting?()
click to toggle source
# File 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
when_connected(&block)
click to toggle source
# File lib/action_cable/subscription_adapter/redis.rb, line 177 def when_connected(&block) if @subscribed_client block.call else @when_connected << block end end