diff options
-rw-r--r-- | Gemfile | 2 | ||||
-rw-r--r-- | Gemfile.lock | 16 | ||||
-rw-r--r-- | actioncable/actioncable.gemspec | 1 | ||||
-rw-r--r-- | actioncable/lib/action_cable/channel/periodic_timers.rb | 4 | ||||
-rw-r--r-- | actioncable/lib/action_cable/channel/streams.rb | 2 | ||||
-rw-r--r-- | actioncable/lib/action_cable/connection/internal_channel.rb | 4 | ||||
-rw-r--r-- | actioncable/lib/action_cable/process/logging.rb | 9 | ||||
-rw-r--r-- | actioncable/lib/action_cable/server.rb | 4 | ||||
-rw-r--r-- | actioncable/lib/action_cable/server/connections.rb | 8 | ||||
-rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/async.rb | 4 | ||||
-rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/postgresql.rb | 4 | ||||
-rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/redis.rb | 16 | ||||
-rw-r--r-- | actioncable/test/channel/periodic_timers_test.rb | 2 | ||||
-rw-r--r-- | actioncable/test/channel/stream_test.rb | 22 | ||||
-rw-r--r-- | actioncable/test/connection/base_test.rb | 13 | ||||
-rw-r--r-- | actioncable/test/subscription_adapter/common.rb | 2 | ||||
-rw-r--r-- | actioncable/test/test_helper.rb | 26 |
17 files changed, 70 insertions, 69 deletions
@@ -66,6 +66,8 @@ group :cable do gem 'em-hiredis', require: false gem 'redis', require: false + + gem 'faye-websocket', github: "matthewd/faye-websocket-ruby", branch: "no-em-concept", require: false end # Add your own local bundler stuff. diff --git a/Gemfile.lock b/Gemfile.lock index 0789c59905..3467c23d6f 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -20,6 +20,16 @@ GIT redis-namespace GIT + remote: git://github.com/matthewd/faye-websocket-ruby.git + revision: f608bb57844b91b397817b06ab6bf744324010ab + branch: no-em-concept + specs: + faye-websocket (0.10.2) + concurrent-ruby (~> 1.0) + nio4r (~> 1.2) + websocket-driver (>= 0.5.1) + +GIT remote: git://github.com/sass/sass.git revision: bce9509f396225d721501ea1070a6871b708abb1 branch: stable @@ -32,7 +42,6 @@ PATH actioncable (5.0.0.beta1) actionpack (= 5.0.0.beta1) coffee-rails (~> 4.1.0) - eventmachine (~> 1.0) faye-websocket (~> 0.10.0) websocket-driver (~> 0.6.1) actionmailer (5.0.0.beta1) @@ -145,9 +154,6 @@ GEM erubis (2.7.0) eventmachine (1.0.9.1) execjs (2.6.0) - faye-websocket (0.10.2) - eventmachine (>= 0.12.0) - websocket-driver (>= 0.5.1) ffi (1.9.10) ffi (1.9.10-x64-mingw32) ffi (1.9.10-x86-mingw32) @@ -185,6 +191,7 @@ GEM mysql2 (0.4.2) mysql2 (0.4.2-x64-mingw32) mysql2 (0.4.2-x86-mingw32) + nio4r (1.2.0) nokogiri (1.6.7.1) mini_portile2 (~> 2.0.0.rc2) nokogiri (1.6.7.1-x64-mingw32) @@ -307,6 +314,7 @@ DEPENDENCIES delayed_job delayed_job_active_record em-hiredis + faye-websocket! jquery-rails json kindlerb (= 0.1.1) 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 |