Class: ActionCable::SubscriptionAdapter::Redis::Listener
Relationships & Source Files | |
Namespace Children | |
Classes:
| |
Super Chains via Extension / Inclusion / Inheritance | |
Class Chain:
|
|
Instance Chain:
|
|
Inherits: |
ActionCable::SubscriptionAdapter::SubscriberMap
|
Defined in: | actioncable/lib/action_cable/subscription_adapter/redis.rb |
Constant Summary
-
ConnectionError =
# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 210RedisClient::ConnectionError
Class Method Summary
Instance Method Summary
Constructor Details
.new(adapter, config_options, event_loop) ⇒ Listener
# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 65
def initialize(adapter, , 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 = .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 Method Details
#add_channel(channel, on_success)
[ GitHub ]# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 138
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
[ GitHub ]# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 152
def invoke_callback(*) @event_loop.post { super } end
#listen(conn)
[ GitHub ]# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 86
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. do |chan, | broadcast(chan, ) 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 146
def remove_channel(channel) @subscription_lock.synchronize do when_connected { @subscribed_client.unsubscribe(channel) } end end
#shutdown
[ GitHub ]# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 125
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