Using a hacked faye-websocket, drop EventMachine

15 files changed, 56 insertions, 65 deletions

diff --git a/actioncable/actioncable.gemspec b/actioncable/actioncable.gemspec
index a36acc8f6f..847fcc71c3 100644
--- a/actioncable/actioncable.gemspec
+++ b/actioncable/actioncable.gemspec
@@ -21,7 +21,6 @@ Gem::Specification.new do |s|
   s.add_dependency 'actionpack', version
 
   s.add_dependency 'coffee-rails',     '~> 4.1.0'
-  s.add_dependency 'eventmachine',     '~> 1.0'
   s.add_dependency 'faye-websocket',   '~> 0.10.0'
   s.add_dependency 'websocket-driver', '~> 0.6.1'
 
diff --git a/actioncable/lib/action_cable/channel/periodic_timers.rb b/actioncable/lib/action_cable/channel/periodic_timers.rb
index 7f0fb37afc..56597d02d7 100644
--- a/actioncable/lib/action_cable/channel/periodic_timers.rb
+++ b/actioncable/lib/action_cable/channel/periodic_timers.rb
@@ -27,14 +27,14 @@ module ActionCable
 
         def start_periodic_timers
           self.class.periodic_timers.each do |callback, options|
-            active_periodic_timers << EventMachine::PeriodicTimer.new(options[:every]) do
+            active_periodic_timers << Concurrent::TimerTask.new(execution_interval: options[:every]) do
               connection.worker_pool.async_run_periodic_timer(self, callback)
             end
           end
         end
 
         def stop_periodic_timers
-          active_periodic_timers.each { |timer| timer.cancel }
+          active_periodic_timers.each { |timer| timer.shutdown }
         end
     end
   end
diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb
index e2876ef6fa..a26373e387 100644
--- a/actioncable/lib/action_cable/channel/streams.rb
+++ b/actioncable/lib/action_cable/channel/streams.rb
@@ -75,7 +75,7 @@ module ActionCable
         callback ||= default_stream_callback(broadcasting)
         streams << [ broadcasting, callback ]
 
-        EM.next_tick do
+        Concurrent.global_io_executor.post do
           pubsub.subscribe(broadcasting, callback, lambda do
             transmit_subscription_confirmation
             logger.info "#{self.class.name} is streaming from #{broadcasting}"
diff --git a/actioncable/lib/action_cable/connection/internal_channel.rb b/actioncable/lib/action_cable/connection/internal_channel.rb
index 54ed7672d2..27826792b3 100644
--- a/actioncable/lib/action_cable/connection/internal_channel.rb
+++ b/actioncable/lib/action_cable/connection/internal_channel.rb
@@ -15,14 +15,14 @@ module ActionCable
             @_internal_subscriptions ||= []
             @_internal_subscriptions << [ internal_channel, callback ]
 
-            EM.next_tick { pubsub.subscribe(internal_channel, callback) }
+            Concurrent.global_io_executor.post { pubsub.subscribe(internal_channel, callback) }
             logger.info "Registered connection (#{connection_identifier})"
           end
         end
 
         def unsubscribe_from_internal_channel
           if @_internal_subscriptions.present?
-            @_internal_subscriptions.each { |channel, callback| EM.next_tick { pubsub.unsubscribe(channel, callback) } }
+            @_internal_subscriptions.each { |channel, callback| Concurrent.global_io_executor.post { pubsub.unsubscribe(channel, callback) } }
           end
         end
 
diff --git a/actioncable/lib/action_cable/process/logging.rb b/actioncable/lib/action_cable/process/logging.rb
index dce637b3ca..ebb8990891 100644
--- a/actioncable/lib/action_cable/process/logging.rb
+++ b/actioncable/lib/action_cable/process/logging.rb
@@ -1,7 +1,8 @@
 require 'action_cable/server'
-require 'eventmachine'
 
-EM.error_handler do |e|
-  puts "Error raised inside the event loop: #{e.message}"
-  puts e.backtrace.join("\n")
+if defined?(::EventMachine)
+  EventMachine.error_handler do |e|
+    puts "Error raised inside the event loop: #{e.message}"
+    puts e.backtrace.join("\n")
+  end
 end
diff --git a/actioncable/lib/action_cable/server.rb b/actioncable/lib/action_cable/server.rb
index a2a89d5f1e..bd6a3826a3 100644
--- a/actioncable/lib/action_cable/server.rb
+++ b/actioncable/lib/action_cable/server.rb
@@ -1,7 +1,3 @@
-require 'eventmachine'
-EventMachine.epoll  if EventMachine.epoll?
-EventMachine.kqueue if EventMachine.kqueue?
-
 module ActionCable
   module Server
     extend ActiveSupport::Autoload
diff --git a/actioncable/lib/action_cable/server/connections.rb b/actioncable/lib/action_cable/server/connections.rb
index 47dcea8c20..8671dd5ebd 100644
--- a/actioncable/lib/action_cable/server/connections.rb
+++ b/actioncable/lib/action_cable/server/connections.rb
@@ -22,11 +22,9 @@ module ActionCable
       # then can't rely on being able to receive and send to it. So there's a 3 second heartbeat running on all connections. If the beat fails, we automatically
       # disconnect.
       def setup_heartbeat_timer
-        EM.next_tick do
-          @heartbeat_timer ||= EventMachine.add_periodic_timer(BEAT_INTERVAL) do
-            EM.next_tick { connections.map(&:beat) }
-          end
-        end
+        @heartbeat_timer ||= Concurrent::TimerTask.new(execution_interval: BEAT_INTERVAL) do
+          Concurrent.global_io_executor.post { connections.map(&:beat) }
+        end.tap(&:execute)
       end
 
       def open_connections_statistics
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
diff --git a/actioncable/test/channel/periodic_timers_test.rb b/actioncable/test/channel/periodic_timers_test.rb
index 1590a12f09..64f0247cd6 100644
--- a/actioncable/test/channel/periodic_timers_test.rb
+++ b/actioncable/test/channel/periodic_timers_test.rb
@@ -31,7 +31,7 @@ class ActionCable::Channel::PeriodicTimersTest < ActiveSupport::TestCase
   end
 
   test "timer start and stop" do
-    EventMachine::PeriodicTimer.expects(:new).times(2).returns(true)
+    Concurrent::TimerTask.expects(:new).times(2).returns(true)
     channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
 
     channel.expects(:stop_periodic_timers).once
diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb
index 3fa2b291b7..947efd96d4 100644
--- a/actioncable/test/channel/stream_test.rb
+++ b/actioncable/test/channel/stream_test.rb
@@ -31,9 +31,7 @@ class ActionCable::Channel::StreamTest < ActionCable::TestCase
   test "stream_for" do
     run_in_eventmachine do
       connection = TestConnection.new
-      EM.next_tick do
-        connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
-      end
+      connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:channel:stream_test:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) }
 
       channel = ChatChannel.new connection, ""
       channel.stream_for Room.new(1)
@@ -41,39 +39,35 @@ class ActionCable::Channel::StreamTest < ActionCable::TestCase
   end
 
   test "stream_from subscription confirmation" do
-    EM.run do
+    run_in_eventmachine do
       connection = TestConnection.new
 
       ChatChannel.new connection, "{id: 1}", { id: 1 }
       assert_nil connection.last_transmission
 
-      EM::Timer.new(0.1) do
-        expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription"
-        connection.transmit(expected)
+      wait_for_async
 
-        assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation within 0.1s"
+      expected = ActiveSupport::JSON.encode "identifier" => "{id: 1}", "type" => "confirm_subscription"
+      connection.transmit(expected)
 
-        EM.run_deferred_callbacks
-        EM.stop
-      end
+      assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation within 0.1s"
     end
   end
 
   test "subscription confirmation should only be sent out once" do
-    EM.run do
+    run_in_eventmachine do
       connection = TestConnection.new
 
       channel = ChatChannel.new connection, "test_channel"
       channel.send_confirmation
       channel.send_confirmation
 
-      EM.run_deferred_callbacks
+      wait_for_async
 
       expected = ActiveSupport::JSON.encode "identifier" => "test_channel", "type" => "confirm_subscription"
       assert_equal expected, connection.last_transmission, "Did not receive subscription confirmation"
 
       assert_equal 1, connection.transmissions.size
-      EM.stop
     end
   end
 
diff --git a/actioncable/test/connection/base_test.rb b/actioncable/test/connection/base_test.rb
index 182562db82..579578d0a7 100644
--- a/actioncable/test/connection/base_test.rb
+++ b/actioncable/test/connection/base_test.rb
@@ -37,6 +37,8 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
       connection.process
 
       assert connection.websocket.possible?
+
+      wait_for_async
       assert connection.websocket.alive?
     end
   end
@@ -58,11 +60,10 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
       connection.websocket.expects(:transmit).with(regexp_matches(/\_ping/))
       connection.message_buffer.expects(:process!)
 
-      # Allow EM to run on_open callback
-      EM.next_tick do
-        assert_equal [ connection ], @server.connections
-        assert connection.connected
-      end
+      wait_for_async
+
+      assert_equal [ connection ], @server.connections
+      assert connection.connected
     end
   end
 
@@ -72,7 +73,7 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
       connection.process
 
       # Setup the connection
-      EventMachine.stubs(:add_periodic_timer).returns(true)
+      Concurrent::TimerTask.stubs(:new).returns(true)
       connection.send :on_open
       assert connection.connected
 
diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb
index d4a13be889..5efc6d6b81 100644
--- a/actioncable/test/subscription_adapter/common.rb
+++ b/actioncable/test/subscription_adapter/common.rb
@@ -24,8 +24,6 @@ module CommonSubscriptionAdapterTest
 
 
     # and now the "real" setup for our test:
-    spawn_eventmachine
-
     server.config.cable = cable_config.with_indifferent_access
 
     adapter_klass = server.config.pubsub_adapter
diff --git a/actioncable/test/test_helper.rb b/actioncable/test/test_helper.rb
index 6636ce078b..8ddbd4e764 100644
--- a/actioncable/test/test_helper.rb
+++ b/actioncable/test/test_helper.rb
@@ -13,28 +13,16 @@ require 'rack/mock'
 # Require all the stubs and models
 Dir[File.dirname(__FILE__) + '/stubs/*.rb'].each {|file| require file }
 
-require 'faye/websocket'
-class << Faye::WebSocket
-  remove_method :ensure_reactor_running
-
-  # We don't want Faye to start the EM reactor in tests because it makes testing much harder.
-  # We want to be able to start and stop EM loop in tests to make things simpler.
-  def ensure_reactor_running
-    # no-op
-  end
-end
-
 class ActionCable::TestCase < ActiveSupport::TestCase
-  def run_in_eventmachine
-    EM.run do
-      yield
-
-      EM.run_deferred_callbacks
-      EM.stop
+  def wait_for_async
+    e = Concurrent.global_io_executor
+    until e.completed_task_count == e.scheduled_task_count
+      sleep 0.1
     end
   end
 
-  def spawn_eventmachine
-    Thread.new { EventMachine.run } unless EventMachine.reactor_running?
+  def run_in_eventmachine
+    yield
+    wait_for_async
   end
 end