Merge pull request #23992 from matthewd/em-option

Support faye-websocket + EventMachine as an option

13 files changed, 73 insertions, 17 deletions

diff --git a/actioncable/test/channel/periodic_timers_test.rb b/actioncable/test/channel/periodic_timers_test.rb
index 64f0247cd6..e6f0c14c9d 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
-    Concurrent::TimerTask.expects(:new).times(2).returns(true)
+    @connection.server.event_loop.expects(:timer).times(2).returns(true)
     channel = ChatChannel.new @connection, "{id: 1}", { id: 1 }
 
     channel.expects(:stop_periodic_timers).once
diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb
index 75545993da..5f5c09d1a1 100644
--- a/actioncable/test/client_test.rb
+++ b/actioncable/test/client_test.rb
@@ -8,8 +8,8 @@ require 'faye/websocket'
 require 'json'
 
 class ClientTest < ActionCable::TestCase
-  WAIT_WHEN_EXPECTING_EVENT = 3
-  WAIT_WHEN_NOT_EXPECTING_EVENT = 0.2
+  WAIT_WHEN_EXPECTING_EVENT = 8
+  WAIT_WHEN_NOT_EXPECTING_EVENT = 0.5
 
   def setup
     ActionCable.instance_variable_set(:@server, nil)
@@ -17,6 +17,7 @@ class ClientTest < ActionCable::TestCase
     server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN }
 
     server.config.cable = { adapter: 'async' }.with_indifferent_access
+    server.config.use_faye = ENV['FAYE'].present?
 
     # and now the "real" setup for our test:
     server.config.disable_request_forgery_protection = true
diff --git a/actioncable/test/connection/authorization_test.rb b/actioncable/test/connection/authorization_test.rb
index 87d0e79ef3..a0506cb9c0 100644
--- a/actioncable/test/connection/authorization_test.rb
+++ b/actioncable/test/connection/authorization_test.rb
@@ -20,7 +20,7 @@ class ActionCable::Connection::AuthorizationTest < ActionCable::TestCase
       server.config.allowed_request_origins = %w( http://rubyonrails.com )
 
       env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket',
-        'HTTP_ORIGIN' => 'http://rubyonrails.com'
+        'HTTP_HOST' => 'localhost', 'HTTP_ORIGIN' => 'http://rubyonrails.com'
 
       connection = Connection.new(server, env)
       connection.websocket.expects(:close)
diff --git a/actioncable/test/connection/base_test.rb b/actioncable/test/connection/base_test.rb
index fb83bd7d77..d7e1041e68 100644
--- a/actioncable/test/connection/base_test.rb
+++ b/actioncable/test/connection/base_test.rb
@@ -1,5 +1,6 @@
 require 'test_helper'
 require 'stubs/test_server'
+require 'active_support/core_ext/object/json'
 
 class ActionCable::Connection::BaseTest < ActionCable::TestCase
   class Connection < ActionCable::Connection::Base
@@ -73,7 +74,7 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
       connection.process
 
       # Setup the connection
-      Concurrent::TimerTask.stubs(:new).returns(true)
+      connection.server.stubs(:timer).returns(true)
       connection.send :handle_open
       assert connection.connected
 
@@ -119,7 +120,7 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
       env = Rack::MockRequest.env_for(
         "/test",
         { 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket',
-          'HTTP_ORIGIN' => 'http://rubyonrails.org', 'rack.hijack' => CallMeMaybe.new }
+          'HTTP_HOST' => 'localhost', 'HTTP_ORIGIN' => 'http://rubyonrails.org', 'rack.hijack' => CallMeMaybe.new }
       )
 
       connection = ActionCable::Connection::Base.new(@server, env)
@@ -131,7 +132,7 @@ class ActionCable::Connection::BaseTest < ActionCable::TestCase
   private
     def open_connection
       env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket',
-        'HTTP_ORIGIN' => 'http://rubyonrails.com'
+        'HTTP_HOST' => 'localhost', 'HTTP_ORIGIN' => 'http://rubyonrails.com'
 
       Connection.new(@server, env)
     end
diff --git a/actioncable/test/connection/cross_site_forgery_test.rb b/actioncable/test/connection/cross_site_forgery_test.rb
index a29f65fb97..2d516b0533 100644
--- a/actioncable/test/connection/cross_site_forgery_test.rb
+++ b/actioncable/test/connection/cross_site_forgery_test.rb
@@ -76,6 +76,6 @@ class ActionCable::Connection::CrossSiteForgeryTest < ActionCable::TestCase
 
     def env_for_origin(origin)
       Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket', 'SERVER_NAME' => HOST,
-        'HTTP_ORIGIN' => origin
+        'HTTP_HOST' => HOST, 'HTTP_ORIGIN' => origin
     end
 end
diff --git a/actioncable/test/connection/identifier_test.rb b/actioncable/test/connection/identifier_test.rb
index 1019ad541e..c3d5f1f90b 100644
--- a/actioncable/test/connection/identifier_test.rb
+++ b/actioncable/test/connection/identifier_test.rb
@@ -64,7 +64,7 @@ class ActionCable::Connection::IdentifierTest < ActionCable::TestCase
     end
 
     def open_connection(server:)
-      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      env = Rack::MockRequest.env_for "/test", 'HTTP_HOST' => 'localhost', 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
       @connection = Connection.new(server, env)
 
       @connection.process
diff --git a/actioncable/test/connection/multiple_identifiers_test.rb b/actioncable/test/connection/multiple_identifiers_test.rb
index e9bb4e6d7f..484e73bb30 100644
--- a/actioncable/test/connection/multiple_identifiers_test.rb
+++ b/actioncable/test/connection/multiple_identifiers_test.rb
@@ -28,7 +28,7 @@ class ActionCable::Connection::MultipleIdentifiersTest < ActionCable::TestCase
     end
 
     def open_connection(server:)
-      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      env = Rack::MockRequest.env_for "/test", 'HTTP_HOST' => 'localhost', 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
       @connection = Connection.new(server, env)
 
       @connection.process
diff --git a/actioncable/test/connection/string_identifier_test.rb b/actioncable/test/connection/string_identifier_test.rb
index 9d0bda83ef..eca0c31060 100644
--- a/actioncable/test/connection/string_identifier_test.rb
+++ b/actioncable/test/connection/string_identifier_test.rb
@@ -30,7 +30,7 @@ class ActionCable::Connection::StringIdentifierTest < ActionCable::TestCase
     end
 
     def open_connection
-      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      env = Rack::MockRequest.env_for "/test", 'HTTP_HOST' => 'localhost', 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
       @connection = Connection.new(@server, env)
 
       @connection.process
diff --git a/actioncable/test/connection/subscriptions_test.rb b/actioncable/test/connection/subscriptions_test.rb
index 68a79c6e4c..f91597f567 100644
--- a/actioncable/test/connection/subscriptions_test.rb
+++ b/actioncable/test/connection/subscriptions_test.rb
@@ -107,7 +107,7 @@ class ActionCable::Connection::SubscriptionsTest < ActionCable::TestCase
     end
 
     def setup_connection
-      env = Rack::MockRequest.env_for "/test", 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
+      env = Rack::MockRequest.env_for "/test", 'HTTP_HOST' => 'localhost', 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket'
       @connection = Connection.new(@server, env)
 
       @subscriptions = ActionCable::Connection::Subscriptions.new(@connection)
diff --git a/actioncable/test/stubs/test_connection.rb b/actioncable/test/stubs/test_connection.rb
index da98201900..8ba284fdc6 100644
--- a/actioncable/test/stubs/test_connection.rb
+++ b/actioncable/test/stubs/test_connection.rb
@@ -1,18 +1,19 @@
 require 'stubs/user'
 
 class TestConnection
-  attr_reader :identifiers, :logger, :current_user, :transmissions
+  attr_reader :identifiers, :logger, :current_user, :server, :transmissions
 
   def initialize(user = User.new("lifo"))
     @identifiers = [ :current_user ]
 
     @current_user = user
     @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new)
+    @server = TestServer.new
     @transmissions = []
   end
 
   def pubsub
-    SuccessAdapter.new(TestServer.new)
+    SuccessAdapter.new(server)
   end
 
   def transmit(data)
diff --git a/actioncable/test/stubs/test_server.rb b/actioncable/test/stubs/test_server.rb
index 5916cf1e83..9e860825f3 100644
--- a/actioncable/test/stubs/test_server.rb
+++ b/actioncable/test/stubs/test_server.rb
@@ -8,14 +8,24 @@ class TestServer
   def initialize
     @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new)
     @config = OpenStruct.new(log_tags: [], subscription_adapter: SuccessAdapter)
+    @config.use_faye = ENV['FAYE'].present?
+    @config.client_socket_class = if @config.use_faye
+                                    ActionCable::Connection::FayeClientSocket
+                                  else
+                                    ActionCable::Connection::ClientSocket
+                                  end
   end
 
   def pubsub
     @config.subscription_adapter.new(self)
   end
 
-  def stream_event_loop
-    @stream_event_loop ||= ActionCable::Connection::StreamEventLoop.new
+  def event_loop
+    @event_loop ||= if @config.use_faye
+                      ActionCable::Connection::FayeEventLoop.new
+                    else
+                      ActionCable::Connection::StreamEventLoop.new
+                    end
   end
 
   def worker_pool
diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb
index b31c2aa36c..82f0abbbf3 100644
--- a/actioncable/test/subscription_adapter/common.rb
+++ b/actioncable/test/subscription_adapter/common.rb
@@ -11,6 +11,7 @@ module CommonSubscriptionAdapterTest
   def setup
     server = ActionCable::Server::Base.new
     server.config.cable = cable_config.with_indifferent_access
+    server.config.use_faye = ENV['FAYE'].present?
 
     adapter_klass = server.config.pubsub_adapter
 
diff --git a/actioncable/test/test_helper.rb b/actioncable/test/test_helper.rb
index 797e7786d1..030362d512 100644
--- a/actioncable/test/test_helper.rb
+++ b/actioncable/test/test_helper.rb
@@ -11,7 +11,41 @@ require 'active_support/core_ext/hash/indifferent_access'
 # Require all the stubs and models
 Dir[File.dirname(__FILE__) + '/stubs/*.rb'].each {|file| require file }
 
-class ActionCable::TestCase < ActiveSupport::TestCase
+if ENV['FAYE'].present?
+  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
+end
+
+module EventMachineConcurrencyHelpers
+  def wait_for_async
+    EM.run_deferred_callbacks
+  end
+
+  def run_in_eventmachine
+    failure = nil
+    EM.run do
+      begin
+        yield
+      rescue => ex
+        failure = ex
+      ensure
+        wait_for_async
+        EM.stop if EM.reactor_running?
+      end
+    end
+    raise failure if failure
+  end
+end
+
+module ConcurrentRubyConcurrencyHelpers
   def wait_for_async
     e = Concurrent.global_io_executor
     until e.completed_task_count == e.scheduled_task_count
@@ -24,3 +58,11 @@ class ActionCable::TestCase < ActiveSupport::TestCase
     wait_for_async
   end
 end
+
+class ActionCable::TestCase < ActiveSupport::TestCase
+  if ENV['FAYE'].present?
+    include EventMachineConcurrencyHelpers
+  else
+    include ConcurrentRubyConcurrencyHelpers
+  end
+end