Revert "Revert "Eliminate the EventMachine dependency""

8 files changed, 32 insertions, 51 deletions

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..e2b017a9a1 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
@@ -53,16 +55,15 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
   test "on connection open" do
     run_in_eventmachine do
       connection = open_connection
-      connection.process
 
       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
+      connection.process
+      wait_for_async
+
+      assert_equal [ connection ], @server.connections
+      assert connection.connected
     end
   end
 
@@ -72,12 +73,12 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
       connection.process
 
       # Setup the connection
-      EventMachine.stubs(:add_periodic_timer).returns(true)
-      connection.send :on_open
+      Concurrent::TimerTask.stubs(:new).returns(true)
+      connection.send :handle_open
       assert connection.connected
 
       connection.subscriptions.expects(:unsubscribe_from_all)
-      connection.send :on_close
+      connection.send :handle_close
 
       assert ! connection.connected
       assert_equal [], @server.connections
diff --git a/actioncable/test/connection/identifier_test.rb b/actioncable/test/connection/identifier_test.rb
index a110dfdee0..1019ad541e 100644
--- a/actioncable/test/connection/identifier_test.rb
+++ b/actioncable/test/connection/identifier_test.rb
@@ -68,10 +68,10 @@ class ActionCable::Connection::IdentifierTest < ActionCable::TestCase
       @connection = Connection.new(server, env)
 
       @connection.process
-      @connection.send :on_open
+      @connection.send :handle_open
     end
 
     def close_connection
-      @connection.send :on_close
+      @connection.send :handle_close
     end
 end
diff --git a/actioncable/test/connection/multiple_identifiers_test.rb b/actioncable/test/connection/multiple_identifiers_test.rb
index 55a9f96cb3..e9bb4e6d7f 100644
--- a/actioncable/test/connection/multiple_identifiers_test.rb
+++ b/actioncable/test/connection/multiple_identifiers_test.rb
@@ -32,10 +32,10 @@ class ActionCable::Connection::MultipleIdentifiersTest < ActionCable::TestCase
       @connection = Connection.new(server, env)
 
       @connection.process
-      @connection.send :on_open
+      @connection.send :handle_open
     end
 
     def close_connection
-      @connection.send :on_close
+      @connection.send :handle_close
     end
 end
diff --git a/actioncable/test/stubs/test_server.rb b/actioncable/test/stubs/test_server.rb
index 6e6541a952..56d132b30a 100644
--- a/actioncable/test/stubs/test_server.rb
+++ b/actioncable/test/stubs/test_server.rb
@@ -14,6 +14,7 @@ class TestServer
     @config.subscription_adapter.new(self)
   end
 
-  def send_async
+  def stream_event_loop
+    @stream_event_loop ||= ActionCable::Connection::StreamEventLoop.new
   end
 end
diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb
index d4a13be889..361858784e 100644
--- a/actioncable/test/subscription_adapter/common.rb
+++ b/actioncable/test/subscription_adapter/common.rb
@@ -1,7 +1,6 @@
 require 'test_helper'
 require 'concurrent'
 
-require 'action_cable/process/logging'
 require 'active_support/core_ext/hash/indifferent_access'
 require 'pathname'
 
@@ -24,8 +23,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