diff options
Diffstat (limited to 'actioncable')
22 files changed, 181 insertions, 259 deletions
diff --git a/actioncable/CHANGELOG.md b/actioncable/CHANGELOG.md index 2e6c2f2afc..137c88d91b 100644 --- a/actioncable/CHANGELOG.md +++ b/actioncable/CHANGELOG.md @@ -1,3 +1,11 @@ +* Prevent race where the client could receive and act upon a + subscription confirmation before the channel's `subscribed` method + completed. + + Fixes #25381. + + *Vladimir Dementyev* + * Buffer writes to websocket connections, to avoid blocking threads that could be doing more useful things. diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb index 2e589a2cfa..a866044f95 100644 --- a/actioncable/lib/action_cable/channel/base.rb +++ b/actioncable/lib/action_cable/channel/base.rb @@ -144,13 +144,14 @@ module ActionCable # When a channel is streaming via pubsub, we want to delay the confirmation # transmission until pubsub subscription is confirmed. - @defer_subscription_confirmation = false + # + # The counter starts at 1 because it's awaiting a call to #subscribe_to_channel + @defer_subscription_confirmation_counter = Concurrent::AtomicFixnum.new(1) @reject_subscription = nil @subscription_confirmation_sent = nil delegate_connection_identifiers - subscribe_to_channel end # Extract the action name from the passed data and process it via the channel. The process will ensure @@ -169,6 +170,17 @@ module ActionCable end end + # This method is called after subscription has been added to the connection + # and confirms or rejects the subscription. + def subscribe_to_channel + run_callbacks :subscribe do + subscribed + end + + reject_subscription if subscription_rejected? + ensure_confirmation_sent + end + # Called by the cable connection when it's cut, so the channel has a chance to cleanup with callbacks. # This method is not intended to be called directly by the user. Instead, overwrite the #unsubscribed callback. def unsubscribe_from_channel # :nodoc: @@ -201,12 +213,18 @@ module ActionCable end end + def ensure_confirmation_sent + return if subscription_rejected? + @defer_subscription_confirmation_counter.decrement + transmit_subscription_confirmation unless defer_subscription_confirmation? + end + def defer_subscription_confirmation! - @defer_subscription_confirmation = true + @defer_subscription_confirmation_counter.increment end def defer_subscription_confirmation? - @defer_subscription_confirmation + @defer_subscription_confirmation_counter.value > 0 end def subscription_confirmation_sent? @@ -230,18 +248,6 @@ module ActionCable end end - def subscribe_to_channel - run_callbacks :subscribe do - subscribed - end - - if subscription_rejected? - reject_subscription - else - transmit_subscription_confirmation unless defer_subscription_confirmation? - end - end - def extract_action(data) (data["action"].presence || :receive).to_sym end diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb index 13deb62662..dbba333353 100644 --- a/actioncable/lib/action_cable/channel/streams.rb +++ b/actioncable/lib/action_cable/channel/streams.rb @@ -84,7 +84,7 @@ module ActionCable connection.server.event_loop.post do pubsub.subscribe(broadcasting, handler, lambda do - transmit_subscription_confirmation + ensure_confirmation_sent logger.info "#{self.class.name} is streaming from #{broadcasting}" end) end diff --git a/actioncable/lib/action_cable/connection.rb b/actioncable/lib/action_cable/connection.rb index 5f813cf8e0..902efb07e2 100644 --- a/actioncable/lib/action_cable/connection.rb +++ b/actioncable/lib/action_cable/connection.rb @@ -8,8 +8,6 @@ module ActionCable autoload :ClientSocket autoload :Identification autoload :InternalChannel - autoload :FayeClientSocket - autoload :FayeEventLoop autoload :MessageBuffer autoload :Stream autoload :StreamEventLoop diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb index 4c7fcc1434..06f4f5edd3 100644 --- a/actioncable/lib/action_cable/connection/base.rb +++ b/actioncable/lib/action_cable/connection/base.rb @@ -57,7 +57,7 @@ module ActionCable @worker_pool = server.worker_pool @logger = new_tagged_logger - @websocket = ActionCable::Connection::WebSocket.new(env, self, event_loop, server.config.client_socket_class) + @websocket = ActionCable::Connection::WebSocket.new(env, self, event_loop) @subscriptions = ActionCable::Connection::Subscriptions.new(self) @message_buffer = ActionCable::Connection::MessageBuffer.new(self) diff --git a/actioncable/lib/action_cable/connection/faye_client_socket.rb b/actioncable/lib/action_cable/connection/faye_client_socket.rb deleted file mode 100644 index 06e92c5d52..0000000000 --- a/actioncable/lib/action_cable/connection/faye_client_socket.rb +++ /dev/null @@ -1,48 +0,0 @@ -require "faye/websocket" - -module ActionCable - module Connection - class FayeClientSocket - def initialize(env, event_target, stream_event_loop, protocols) - @env = env - @event_target = event_target - @protocols = protocols - - @faye = nil - end - - def alive? - @faye && @faye.ready_state == Faye::WebSocket::API::OPEN - end - - def transmit(data) - connect - @faye.send data - end - - def close - @faye && @faye.close - end - - def protocol - @faye && @faye.protocol - end - - def rack_response - connect - @faye.rack_response - end - - private - def connect - return if @faye - @faye = Faye::WebSocket.new(@env, @protocols) - - @faye.on(:open) { |event| @event_target.on_open } - @faye.on(:message) { |event| @event_target.on_message(event.data) } - @faye.on(:close) { |event| @event_target.on_close(event.reason, event.code) } - @faye.on(:error) { |event| @event_target.on_error(event.message) } - end - end - end -end diff --git a/actioncable/lib/action_cable/connection/faye_event_loop.rb b/actioncable/lib/action_cable/connection/faye_event_loop.rb deleted file mode 100644 index cfbe26ee6a..0000000000 --- a/actioncable/lib/action_cable/connection/faye_event_loop.rb +++ /dev/null @@ -1,44 +0,0 @@ -require "thread" - -require "eventmachine" -EventMachine.epoll if EventMachine.epoll? -EventMachine.kqueue if EventMachine.kqueue? - -module ActionCable - module Connection - class FayeEventLoop - @@mutex = Mutex.new - - def timer(interval, &block) - ensure_reactor_running - EMTimer.new(::EM::PeriodicTimer.new(interval, &block)) - end - - def post(task = nil, &block) - task ||= block - - ensure_reactor_running - ::EM.next_tick(&task) - end - - private - 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 - - class EMTimer - def initialize(inner) - @inner = inner - end - - def shutdown - @inner.cancel - end - end - end - end -end diff --git a/actioncable/lib/action_cable/connection/subscriptions.rb b/actioncable/lib/action_cable/connection/subscriptions.rb index 9060183249..00511aead5 100644 --- a/actioncable/lib/action_cable/connection/subscriptions.rb +++ b/actioncable/lib/action_cable/connection/subscriptions.rb @@ -26,10 +26,14 @@ module ActionCable id_key = data["identifier"] id_options = ActiveSupport::JSON.decode(id_key).with_indifferent_access + return if subscriptions.key?(id_key) + subscription_klass = id_options[:channel].safe_constantize if subscription_klass && ActionCable::Channel::Base >= subscription_klass - subscriptions[id_key] ||= subscription_klass.new(connection, id_key, id_options) + subscription = subscription_klass.new(connection, id_key, id_options) + subscriptions[id_key] = subscription + subscription.subscribe_to_channel else logger.error "Subscription class not found: #{id_options[:channel].inspect}" end diff --git a/actioncable/lib/action_cable/connection/web_socket.rb b/actioncable/lib/action_cable/connection/web_socket.rb index 52d8daad4b..382141b89f 100644 --- a/actioncable/lib/action_cable/connection/web_socket.rb +++ b/actioncable/lib/action_cable/connection/web_socket.rb @@ -4,8 +4,8 @@ module ActionCable module Connection # Wrap the real socket to minimize the externally-presented API class WebSocket - def initialize(env, event_target, event_loop, client_socket_class, protocols: ActionCable::INTERNAL[:protocols]) - @websocket = ::WebSocket::Driver.websocket?(env) ? client_socket_class.new(env, event_target, event_loop, protocols) : nil + def initialize(env, event_target, event_loop, protocols: ActionCable::INTERNAL[:protocols]) + @websocket = ::WebSocket::Driver.websocket?(env) ? ClientSocket.new(env, event_target, event_loop, protocols) : nil end def possible? diff --git a/actioncable/lib/action_cable/server/base.rb b/actioncable/lib/action_cable/server/base.rb index dd059a553b..67ada7cc2e 100644 --- a/actioncable/lib/action_cable/server/base.rb +++ b/actioncable/lib/action_cable/server/base.rb @@ -49,7 +49,7 @@ module ActionCable end def event_loop - @event_loop || @mutex.synchronize { @event_loop ||= config.event_loop_class.new } + @event_loop || @mutex.synchronize { @event_loop ||= ActionCable::Connection::StreamEventLoop.new } end # The worker pool is where we run connection callbacks and channel actions. We do as little as possible on the server's main thread. diff --git a/actioncable/lib/action_cable/server/configuration.rb b/actioncable/lib/action_cable/server/configuration.rb index 7153593d4c..dc146f07b0 100644 --- a/actioncable/lib/action_cable/server/configuration.rb +++ b/actioncable/lib/action_cable/server/configuration.rb @@ -4,7 +4,7 @@ module ActionCable # in a Rails config initializer. class Configuration attr_accessor :logger, :log_tags - attr_accessor :use_faye, :connection_class, :worker_pool_size + attr_accessor :connection_class, :worker_pool_size attr_accessor :disable_request_forgery_protection, :allowed_request_origins attr_accessor :cable, :url, :mount_path @@ -35,22 +35,6 @@ module ActionCable adapter = "PostgreSQL" if adapter == "Postgresql" "ActionCable::SubscriptionAdapter::#{adapter}".constantize end - - def event_loop_class - if use_faye - ActionCable::Connection::FayeEventLoop - else - ActionCable::Connection::StreamEventLoop - end - end - - def client_socket_class - if use_faye - ActionCable::Connection::FayeClientSocket - else - ActionCable::Connection::ClientSocket - end - end end end end diff --git a/actioncable/test/channel/base_test.rb b/actioncable/test/channel/base_test.rb index 2bb3214f74..9a3a3581e6 100644 --- a/actioncable/test/channel/base_test.rb +++ b/actioncable/test/channel/base_test.rb @@ -77,11 +77,13 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase @channel = ChatChannel.new @connection, "{id: 1}", id: 1 end - test "should subscribe to a channel on initialize" do + test "should subscribe to a channel" do + @channel.subscribe_to_channel assert_equal 1, @channel.room.id end test "on subscribe callbacks" do + @channel.subscribe_to_channel assert @channel.subscribed end @@ -90,6 +92,8 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase end test "unsubscribing from a channel" do + @channel.subscribe_to_channel + assert @channel.room assert @channel.subscribed? @@ -150,8 +154,13 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase assert_equal expected, @connection.last_transmission end - test "subscription confirmation" do + test "do not send subscription confirmation on initialize" do + assert_nil @connection.last_transmission + end + + test "subscription confirmation on subscribe_to_channel" do expected = { "identifier" => "{id: 1}", "type" => "confirm_subscription" } + @channel.subscribe_to_channel assert_equal expected, @connection.last_transmission end @@ -208,6 +217,8 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase test "notification for transmit_subscription_confirmation" do begin + @channel.subscribe_to_channel + events = [] ActiveSupport::Notifications.subscribe "transmit_subscription_confirmation.action_cable" do |*args| events << ActiveSupport::Notifications::Event.new(*args) diff --git a/actioncable/test/channel/periodic_timers_test.rb b/actioncable/test/channel/periodic_timers_test.rb index 529abb9535..2ee711fd29 100644 --- a/actioncable/test/channel/periodic_timers_test.rb +++ b/actioncable/test/channel/periodic_timers_test.rb @@ -62,6 +62,7 @@ class ActionCable::Channel::PeriodicTimersTest < ActiveSupport::TestCase @connection.server.event_loop.expects(:timer).times(3).returns(stub(shutdown: nil)) channel = ChatChannel.new @connection, "{id: 1}", id: 1 + channel.subscribe_to_channel channel.unsubscribe_from_channel assert_equal [], channel.send(:active_periodic_timers) end diff --git a/actioncable/test/channel/rejection_test.rb b/actioncable/test/channel/rejection_test.rb index faf35ad048..99c4a7603a 100644 --- a/actioncable/test/channel/rejection_test.rb +++ b/actioncable/test/channel/rejection_test.rb @@ -20,6 +20,7 @@ class ActionCable::Channel::RejectionTest < ActiveSupport::TestCase test "subscription rejection" do @connection.expects(:subscriptions).returns mock().tap { |m| m.expects(:remove_subscription).with instance_of(SecretChannel) } @channel = SecretChannel.new @connection, "{id: 1}", id: 1 + @channel.subscribe_to_channel expected = { "identifier" => "{id: 1}", "type" => "reject_subscription" } assert_equal expected, @connection.last_transmission @@ -28,6 +29,7 @@ class ActionCable::Channel::RejectionTest < ActiveSupport::TestCase test "does not execute action if subscription is rejected" do @connection.expects(:subscriptions).returns mock().tap { |m| m.expects(:remove_subscription).with instance_of(SecretChannel) } @channel = SecretChannel.new @connection, "{id: 1}", id: 1 + @channel.subscribe_to_channel expected = { "identifier" => "{id: 1}", "type" => "reject_subscription" } assert_equal expected, @connection.last_transmission diff --git a/actioncable/test/channel/stream_test.rb b/actioncable/test/channel/stream_test.rb index da26f81a5c..31dcde2e29 100644 --- a/actioncable/test/channel/stream_test.rb +++ b/actioncable/test/channel/stream_test.rb @@ -53,6 +53,7 @@ module ActionCable::StreamTests connection = TestConnection.new connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("test_room_1", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } channel = ChatChannel.new connection, "{id: 1}", id: 1 + channel.subscribe_to_channel connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) } channel.unsubscribe_from_channel @@ -64,6 +65,7 @@ module ActionCable::StreamTests connection = TestConnection.new connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("channel", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } channel = SymbolChannel.new connection, "" + channel.subscribe_to_channel connection.expects(:pubsub).returns mock().tap { |m| m.expects(:unsubscribe) } channel.unsubscribe_from_channel @@ -76,6 +78,7 @@ module ActionCable::StreamTests connection.expects(:pubsub).returns mock().tap { |m| m.expects(:subscribe).with("action_cable:stream_tests:chat:Room#1-Campfire", kind_of(Proc), kind_of(Proc)).returns stub_everything(:pubsub) } channel = ChatChannel.new connection, "" + channel.subscribe_to_channel channel.stream_for Room.new(1) end end @@ -84,7 +87,9 @@ module ActionCable::StreamTests run_in_eventmachine do connection = TestConnection.new - ChatChannel.new connection, "{id: 1}", id: 1 + channel = ChatChannel.new connection, "{id: 1}", id: 1 + channel.subscribe_to_channel + assert_nil connection.last_transmission wait_for_async @@ -114,7 +119,7 @@ module ActionCable::StreamTests end end - require "action_cable/subscription_adapter/inline" + require "action_cable/subscription_adapter/async" class UserCallbackChannel < ActionCable::Channel::Base def subscribed @@ -124,9 +129,16 @@ module ActionCable::StreamTests end end - class StreamEncodingTest < ActionCable::TestCase + class MultiChatChannel < ActionCable::Channel::Base + def subscribed + stream_from "main_room" + stream_from "test_all_rooms" + end + end + + class StreamFromTest < ActionCable::TestCase setup do - @server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Inline) + @server = TestServer.new(subscription_adapter: ActionCable::SubscriptionAdapter::Async) @server.config.allowed_request_origins = %w( http://rubyonrails.com ) end @@ -153,6 +165,17 @@ module ActionCable::StreamTests end end + test "subscription confirmation should only be sent out once with muptiple stream_from" do + run_in_eventmachine do + connection = open_connection + expected = { "identifier" => { "channel" => MultiChatChannel.name }.to_json, "type" => "confirm_subscription" } + connection.websocket.expects(:transmit).with(expected.to_json) + receive(connection, command: "subscribe", channel: MultiChatChannel.name, identifiers: {}) + + wait_for_async + end + end + private def subscribe_to(connection, identifiers:) receive connection, command: "subscribe", identifiers: identifiers diff --git a/actioncable/test/client_test.rb b/actioncable/test/client_test.rb index 2e3821828f..f6d4ab3202 100644 --- a/actioncable/test/client_test.rb +++ b/actioncable/test/client_test.rb @@ -1,13 +1,38 @@ require "test_helper" require "concurrent" -require "faye/websocket" +require "websocket-client-simple" require "json" require "active_support/hash_with_indifferent_access" +#### +# 😷 Warning suppression 😷 +WebSocket::Frame::Handler::Handler03.prepend Module.new { + def initialize(*) + @application_data_buffer = nil + super + end +} + +WebSocket::Frame::Data.prepend Module.new { + def initialize(*) + @masking_key = nil + super + end +} + +WebSocket::Client::Simple::Client.prepend Module.new { + def initialize(*) + @socket = nil + super + end +} +# +#### + class ClientTest < ActionCable::TestCase - WAIT_WHEN_EXPECTING_EVENT = 8 + WAIT_WHEN_EXPECTING_EVENT = 2 WAIT_WHEN_NOT_EXPECTING_EVENT = 0.5 class EchoChannel < ActionCable::Channel::Base @@ -39,20 +64,9 @@ class ClientTest < ActionCable::TestCase server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } server.config.cable = ActiveSupport::HashWithIndifferentAccess.new(adapter: "async") - server.config.use_faye = ENV["FAYE"].present? # and now the "real" setup for our test: server.config.disable_request_forgery_protection = true - - Thread.new { EventMachine.run } unless EventMachine.reactor_running? - Thread.pass until EventMachine.reactor_running? - - # faye-websocket is warning-rich - @previous_verbose, $VERBOSE = $VERBOSE, nil - end - - def teardown - $VERBOSE = @previous_verbose end def with_puma_server(rack_app = ActionCable.server, port = 3099) @@ -73,44 +87,49 @@ class ClientTest < ActionCable::TestCase attr_reader :pings def initialize(port) - @ws = Faye::WebSocket::Client.new("ws://127.0.0.1:#{port}/") - @messages = Queue.new - @closed = Concurrent::Event.new - @has_messages = Concurrent::Semaphore.new(0) - @pings = 0 - - open = Concurrent::Event.new - error = nil - - @ws.on(:error) do |event| - if open.set? - @messages << RuntimeError.new(event.message) - else - error = event.message - open.set + messages = @messages = Queue.new + closed = @closed = Concurrent::Event.new + has_messages = @has_messages = Concurrent::Semaphore.new(0) + pings = @pings = Concurrent::AtomicFixnum.new(0) + + open = Concurrent::Promise.new + + @ws = WebSocket::Client::Simple.connect("ws://127.0.0.1:#{port}/") do |ws| + ws.on(:error) do |event| + event = RuntimeError.new(event.message) unless event.is_a?(Exception) + + if open.pending? + open.fail(event) + else + messages << event + has_messages.release + end end - end - @ws.on(:open) do |event| - open.set - end + ws.on(:open) do |event| + open.set(true) + end - @ws.on(:message) do |event| - message = JSON.parse(event.data) - if message["type"] == "ping" - @pings += 1 - else - @messages << message - @has_messages.release + ws.on(:message) do |event| + if event.type == :close + closed.set + else + message = JSON.parse(event.data) + if message["type"] == "ping" + pings.increment + else + messages << message + has_messages.release + end + end end - end - @ws.on(:close) do |event| - @closed.set + ws.on(:close) do |event| + closed.set + end end - open.wait(WAIT_WHEN_EXPECTING_EVENT) - raise error if error + open.wait!(WAIT_WHEN_EXPECTING_EVENT) end def read_message @@ -161,13 +180,17 @@ class ClientTest < ActionCable::TestCase end end - def faye_client(port) + def websocket_client(port) SyncClient.new(port) end + def concurrently(enum) + enum.map { |*x| Concurrent::Future.execute { yield(*x) } }.map(&:value!) + end + def test_single_client with_puma_server do |port| - c = faye_client(port) + c = websocket_client(port) assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") assert_equal({ "identifier"=>"{\"channel\":\"ClientTest::EchoChannel\"}", "type"=>"confirm_subscription" }, c.read_message) @@ -179,12 +202,12 @@ class ClientTest < ActionCable::TestCase def test_interacting_clients with_puma_server do |port| - clients = 10.times.map { faye_client(port) } + clients = concurrently(10.times) { websocket_client(port) } barrier_1 = Concurrent::CyclicBarrier.new(clients.size) barrier_2 = Concurrent::CyclicBarrier.new(clients.size) - clients.map { |c| Concurrent::Future.execute { + concurrently(clients) do |c| assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") assert_equal({ "identifier"=>'{"channel":"ClientTest::EchoChannel"}', "type"=>"confirm_subscription" }, c.read_message) @@ -194,38 +217,38 @@ class ClientTest < ActionCable::TestCase c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "bulk", message: "hello") barrier_2.wait WAIT_WHEN_EXPECTING_EVENT assert_equal clients.size, c.read_messages(clients.size).size - } }.each(&:wait!) + end - clients.map { |c| Concurrent::Future.execute { c.close } }.each(&:wait!) + concurrently(clients, &:close) end end def test_many_clients with_puma_server do |port| - clients = 100.times.map { faye_client(port) } + clients = concurrently(100.times) { websocket_client(port) } - clients.map { |c| Concurrent::Future.execute { + concurrently(clients) do |c| assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") assert_equal({ "identifier"=>'{"channel":"ClientTest::EchoChannel"}', "type"=>"confirm_subscription" }, c.read_message) c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "ding", message: "hello") assert_equal({ "identifier"=>'{"channel":"ClientTest::EchoChannel"}', "message"=>{ "dong"=>"hello" } }, c.read_message) - } }.each(&:wait!) + end - clients.map { |c| Concurrent::Future.execute { c.close } }.each(&:wait!) + concurrently(clients, &:close) end end def test_disappearing_client with_puma_server do |port| - c = faye_client(port) + c = websocket_client(port) assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") assert_equal({ "identifier"=>"{\"channel\":\"ClientTest::EchoChannel\"}", "type"=>"confirm_subscription" }, c.read_message) c.send_message command: "message", identifier: JSON.generate(channel: "ClientTest::EchoChannel"), data: JSON.generate(action: "delay", message: "hello") c.close # disappear before write - c = faye_client(port) + c = websocket_client(port) assert_equal({ "type" => "welcome" }, c.read_message) # pop the first welcome message off the stack c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") assert_equal({ "identifier"=>"{\"channel\":\"ClientTest::EchoChannel\"}", "type"=>"confirm_subscription" }, c.read_message) @@ -240,7 +263,7 @@ class ClientTest < ActionCable::TestCase app = ActionCable.server identifier = JSON.generate(channel: "ClientTest::EchoChannel") - c = faye_client(port) + c = websocket_client(port) assert_equal({ "type" => "welcome" }, c.read_message) c.send_message command: "subscribe", identifier: identifier assert_equal({ "identifier"=>"{\"channel\":\"ClientTest::EchoChannel\"}", "type"=>"confirm_subscription" }, c.read_message) @@ -261,7 +284,7 @@ class ClientTest < ActionCable::TestCase def test_server_restart with_puma_server do |port| - c = faye_client(port) + c = websocket_client(port) assert_equal({ "type" => "welcome" }, c.read_message) c.send_message command: "subscribe", identifier: JSON.generate(channel: "ClientTest::EchoChannel") assert_equal({ "identifier"=>"{\"channel\":\"ClientTest::EchoChannel\"}", "type"=>"confirm_subscription" }, c.read_message) diff --git a/actioncable/test/connection/client_socket_test.rb b/actioncable/test/connection/client_socket_test.rb index 5043a63370..dff7fefbfb 100644 --- a/actioncable/test/connection/client_socket_test.rb +++ b/actioncable/test/connection/client_socket_test.rb @@ -33,8 +33,6 @@ class ActionCable::Connection::ClientSocketTest < ActionCable::TestCase end test "delegate socket errors to on_error handler" do - skip if ENV["FAYE"].present? - run_in_eventmachine do connection = open_connection @@ -49,8 +47,6 @@ class ActionCable::Connection::ClientSocketTest < ActionCable::TestCase end test "closes hijacked i/o socket at shutdown" do - skip if ENV["FAYE"].present? - run_in_eventmachine do connection = open_connection diff --git a/actioncable/test/connection/stream_test.rb b/actioncable/test/connection/stream_test.rb index 4128b32f15..36e1d3c095 100644 --- a/actioncable/test/connection/stream_test.rb +++ b/actioncable/test/connection/stream_test.rb @@ -34,8 +34,6 @@ class ActionCable::Connection::StreamTest < ActionCable::TestCase [ EOFError, Errno::ECONNRESET ].each do |closed_exception| test "closes socket on #{closed_exception}" do - skip if ENV["FAYE"].present? - run_in_eventmachine do connection = open_connection diff --git a/actioncable/test/stubs/test_server.rb b/actioncable/test/stubs/test_server.rb index 02be72d0cb..5bf2a151dc 100644 --- a/actioncable/test/stubs/test_server.rb +++ b/actioncable/test/stubs/test_server.rb @@ -10,12 +10,6 @@ class TestServer @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new) @config = OpenStruct.new(log_tags: [], subscription_adapter: subscription_adapter) - @config.use_faye = ENV["FAYE"].present? - @config.client_socket_class = if @config.use_faye - ActionCable::Connection::FayeClientSocket - else - ActionCable::Connection::ClientSocket - end @mutex = Monitor.new end @@ -25,12 +19,8 @@ class TestServer end def event_loop - @event_loop ||= if @config.use_faye - ActionCable::Connection::FayeEventLoop.new - else - ActionCable::Connection::StreamEventLoop.new.tap do |loop| - loop.instance_variable_set(:@executor, Concurrent.global_io_executor) - end + @event_loop ||= ActionCable::Connection::StreamEventLoop.new.tap do |loop| + loop.instance_variable_set(:@executor, Concurrent.global_io_executor) end end diff --git a/actioncable/test/subscription_adapter/common.rb b/actioncable/test/subscription_adapter/common.rb index 1538157995..8cf0b5c684 100644 --- a/actioncable/test/subscription_adapter/common.rb +++ b/actioncable/test/subscription_adapter/common.rb @@ -11,7 +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? + server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } adapter_klass = server.config.pubsub_adapter diff --git a/actioncable/test/subscription_adapter/evented_redis_test.rb b/actioncable/test/subscription_adapter/evented_redis_test.rb index d6ca0e77cb..f316bc46ef 100644 --- a/actioncable/test/subscription_adapter/evented_redis_test.rb +++ b/actioncable/test/subscription_adapter/evented_redis_test.rb @@ -12,6 +12,14 @@ class EventedRedisAdapterTest < ActionCable::TestCase end def teardown + super + + # Ensure EM is shut down before we re-enable warnings + EventMachine.reactor_thread.tap do |thread| + EventMachine.stop + thread.join + end + $VERBOSE = @previous_verbose end diff --git a/actioncable/test/test_helper.rb b/actioncable/test/test_helper.rb index 39855ea252..a47032753b 100644 --- a/actioncable/test/test_helper.rb +++ b/actioncable/test/test_helper.rb @@ -13,41 +13,7 @@ end # Require all the stubs and models Dir[File.dirname(__FILE__) + "/stubs/*.rb"].each { |file| require file } -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 +class ActionCable::TestCase < ActiveSupport::TestCase def wait_for_async wait_for_executor Concurrent.global_io_executor end @@ -56,18 +22,14 @@ module ConcurrentRubyConcurrencyHelpers yield wait_for_async end -end - -class ActionCable::TestCase < ActiveSupport::TestCase - if ENV["FAYE"].present? - include EventMachineConcurrencyHelpers - else - include ConcurrentRubyConcurrencyHelpers - end def wait_for_executor(executor) + # do not wait forever, wait 2s + timeout = 2 until executor.completed_task_count == executor.scheduled_task_count sleep 0.1 + timeout -= 0.1 + raise "Executor could not complete all tasks in 2 seconds" unless timeout > 0 end end end |