Support faye-websocket + EventMachine as an option

3 files changed, 18 insertions, 11 deletions

diff --git a/actioncable/lib/action_cable/subscription_adapter/async.rb b/actioncable/lib/action_cable/subscription_adapter/async.rb
index cca6894289..10b3ac8cd8 100644
--- a/actioncable/lib/action_cable/subscription_adapter/async.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/async.rb
@@ -5,16 +5,21 @@ module ActionCable
     class Async < Inline # :nodoc:
       private
         def new_subscriber_map
-          AsyncSubscriberMap.new
+          AsyncSubscriberMap.new(server.event_loop)
         end
 
         class AsyncSubscriberMap < SubscriberMap
+          def initialize(event_loop)
+            @event_loop = event_loop
+            super()
+          end
+
           def add_subscriber(*)
-            Concurrent.global_io_executor.post { super }
+            @event_loop.post { super }
           end
 
           def invoke_callback(*)
-            Concurrent.global_io_executor.post { super }
+            @event_loop.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 abaeb92e54..66c7852f6e 100644
--- a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
@@ -42,14 +42,15 @@ module ActionCable
 
       private
         def listener
-          @listener || @server.mutex.synchronize { @listener ||= Listener.new(self) }
+          @listener || @server.mutex.synchronize { @listener ||= Listener.new(self, @server.event_loop) }
         end
 
         class Listener < SubscriberMap
-          def initialize(adapter)
+          def initialize(adapter, event_loop)
             super()
 
             @adapter = adapter
+            @event_loop = event_loop
             @queue = Queue.new
 
             @thread = Thread.new do
@@ -68,7 +69,7 @@ module ActionCable
                     case action
                     when :listen
                       pg_conn.exec("LISTEN #{pg_conn.escape_identifier channel}")
-                      Concurrent.global_io_executor << callback if callback
+                      @event_loop.post(&callback) if callback
                     when :unlisten
                       pg_conn.exec("UNLISTEN #{pg_conn.escape_identifier channel}")
                     when :shutdown
@@ -98,7 +99,7 @@ module ActionCable
           end
 
           def invoke_callback(*)
-            Concurrent.global_io_executor.post { super }
+            @event_loop.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 6b4236e7d3..65434f7107 100644
--- a/actioncable/lib/action_cable/subscription_adapter/redis.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb
@@ -38,7 +38,7 @@ module ActionCable
 
       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
@@ -52,10 +52,11 @@ module ActionCable
         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
@@ -84,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
@@ -133,7 +134,7 @@ module ActionCable
           end
 
           def invoke_callback(*)
-            Concurrent.global_io_executor.post { super }
+            @event_loop.post { super }
           end
 
           private