diff options
Diffstat (limited to 'actioncable/lib/action_cable/subscription_adapter/redis.rb')
-rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/redis.rb | 31 |
1 files changed, 18 insertions, 13 deletions
diff --git a/actioncable/lib/action_cable/subscription_adapter/redis.rb b/actioncable/lib/action_cable/subscription_adapter/redis.rb index ba4934a264..62bd284a6b 100644 --- a/actioncable/lib/action_cable/subscription_adapter/redis.rb +++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb @@ -1,7 +1,7 @@ -require 'thread' +require "thread" -gem 'redis', '~> 3.0' -require 'redis' +gem "redis", "~> 3.0" +require "redis" module ActionCable module SubscriptionAdapter @@ -33,25 +33,30 @@ module ActionCable end def redis_connection_for_subscriptions - ::Redis.new(@server.config.cable) + redis_connection end private def listener - @listener || @server.mutex.synchronize { @listener ||= Listener.new(self) } + @listener || @server.mutex.synchronize { @listener ||= Listener.new(self, @server.event_loop) } end def redis_connection_for_broadcasts @redis_connection_for_broadcasts || @server.mutex.synchronize do - @redis_connection_for_broadcasts ||= self.class.redis_connector.call(@server.config.cable) + @redis_connection_for_broadcasts ||= redis_connection end end + def redis_connection + self.class.redis_connector.call(@server.config.cable) + end + class Listener < SubscriberMap - def initialize(adapter) + def initialize(adapter, event_loop) super() @adapter = adapter + @event_loop = event_loop @subscribe_callbacks = Hash.new { |h, k| h[k] = [] } @subscription_lock = Mutex.new @@ -67,7 +72,7 @@ module ActionCable conn.without_reconnect do original_client = conn.client - conn.subscribe('_action_cable_internal') do |on| + conn.subscribe("_action_cable_internal") do |on| on.subscribe do |chan, count| @subscription_lock.synchronize do if count == 1 @@ -80,7 +85,7 @@ module ActionCable if callbacks = @subscribe_callbacks[chan] next_callback = callbacks.shift - Concurrent.global_io_executor << next_callback if next_callback + @event_loop.post(&next_callback) if next_callback @subscribe_callbacks.delete(chan) if callbacks.empty? end end @@ -106,7 +111,7 @@ module ActionCable return if @thread.nil? when_connected do - send_command('unsubscribe') + send_command("unsubscribe") @raw_client = nil end end @@ -118,18 +123,18 @@ module ActionCable @subscription_lock.synchronize do ensure_listener_running @subscribe_callbacks[channel] << on_success - when_connected { send_command('subscribe', channel) } + when_connected { send_command("subscribe", channel) } end end def remove_channel(channel) @subscription_lock.synchronize do - when_connected { send_command('unsubscribe', channel) } + when_connected { send_command("unsubscribe", channel) } end end def invoke_callback(*) - Concurrent.global_io_executor.post { super } + @event_loop.post { super } end private |