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 213RedisClient::ConnectionError
Class Method Summary
Instance Attribute Summary
- #retry_connecting? ⇒ Boolean readonly private
Instance Method Summary
- #add_channel(channel, on_success)
- #invoke_callback
- #listen(conn)
- #remove_channel(channel)
- #shutdown
- #ensure_listener_running private
-
#extract_subscribed_client(conn)
private
See additional method definition at line 243.
- #reset private
- #resubscribe private
- #when_connected(&block) private
::ActionCable::SubscriptionAdapter::SubscriberMap
- Inherited
Constructor Details
.new(adapter, config_options, event_loop) ⇒ Listener
# File 'actioncable/lib/action_cable/subscription_adapter/redis.rb', line 68
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 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.
# 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. 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 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