aboutsummaryrefslogtreecommitdiffstats
path: root/actioncable/lib/action_cable/subscription_adapter
diff options
context:
space:
mode:
Diffstat (limited to 'actioncable/lib/action_cable/subscription_adapter')
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/async.rb4
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/postgresql.rb4
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/redis.rb16
3 files changed, 20 insertions, 4 deletions
diff --git a/actioncable/lib/action_cable/subscription_adapter/async.rb b/actioncable/lib/action_cable/subscription_adapter/async.rb
index 85d4892e4c..c88b03947a 100644
--- a/actioncable/lib/action_cable/subscription_adapter/async.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/async.rb
@@ -10,11 +10,11 @@ module ActionCable
class AsyncSubscriberMap < SubscriberMap
def add_subscriber(*)
- ::EM.next_tick { super }
+ Concurrent.global_io_executor.post { super }
end
def invoke_callback(*)
- ::EM.next_tick { super }
+ Concurrent.global_io_executor.post { super }
end
end
end
diff --git a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
index 78f8aeb599..3ce1bbed68 100644
--- a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
@@ -63,7 +63,7 @@ module ActionCable
case action
when :listen
pg_conn.exec("LISTEN #{pg_conn.escape_identifier channel}")
- ::EM.next_tick(&callback) if callback
+ Concurrent.global_io_executor << callback if callback
when :unlisten
pg_conn.exec("UNLISTEN #{pg_conn.escape_identifier channel}")
when :shutdown
@@ -93,7 +93,7 @@ module ActionCable
end
def invoke_callback(*)
- ::EM.next_tick { super }
+ Concurrent.global_io_executor.post { super }
end
end
end
diff --git a/actioncable/lib/action_cable/subscription_adapter/redis.rb b/actioncable/lib/action_cable/subscription_adapter/redis.rb
index 3b86354621..a035e3988d 100644
--- a/actioncable/lib/action_cable/subscription_adapter/redis.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb
@@ -1,11 +1,18 @@
+require 'thread'
+
gem 'em-hiredis', '~> 0.3.0'
gem 'redis', '~> 3.0'
require 'em-hiredis'
require 'redis'
+EventMachine.epoll if EventMachine.epoll?
+EventMachine.kqueue if EventMachine.kqueue?
+
module ActionCable
module SubscriptionAdapter
class Redis < Base # :nodoc:
+ @@mutex = Mutex.new
+
def broadcast(channel, payload)
redis_connection_for_broadcasts.publish(channel, payload)
end
@@ -27,6 +34,7 @@ module ActionCable
private
def redis_connection_for_subscriptions
+ ensure_reactor_running
@redis_connection_for_subscriptions ||= EM::Hiredis.connect(@server.config.cable[:url]).tap do |redis|
redis.on(:reconnect_failed) do
@logger.info "[ActionCable] Redis reconnect failed."
@@ -37,6 +45,14 @@ module ActionCable
def redis_connection_for_broadcasts
@redis_connection_for_broadcasts ||= ::Redis.new(@server.config.cable)
end
+
+ def ensure_reactor_running
+ return if EventMachine.reactor_running?
+ @@mutex.synchronize do
+ Thread.new { EventMachine.run } unless EventMachine.reactor_running?
+ Thread.pass until EventMachine.reactor_running?
+ end
+ end
end
end
end