Merge pull request #23992 from matthewd/em-option

Support faye-websocket + EventMachine as an option

1 files changed, 5 insertions, 4 deletions

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