diff options
Diffstat (limited to 'actioncable')
12 files changed, 247 insertions, 8 deletions
diff --git a/actioncable/CHANGELOG.md b/actioncable/CHANGELOG.md index fb9e498e49..1ea862b389 100644 --- a/actioncable/CHANGELOG.md +++ b/actioncable/CHANGELOG.md @@ -1,3 +1,7 @@ +* Add ActiveSupport::Notifications to ActionCable::Channel. + + *Matthew Wear* + * Allow channel identifiers with no backslahes/escaping to be accepted by the subscription storer. diff --git a/actioncable/app/assets/javascripts/action_cable/connection.coffee b/actioncable/app/assets/javascripts/action_cable/connection.coffee index 92272cc5b8..3a139acf3a 100644 --- a/actioncable/app/assets/javascripts/action_cable/connection.coffee +++ b/actioncable/app/assets/javascripts/action_cable/connection.coffee @@ -20,7 +20,7 @@ class ActionCable.Connection open: => if @isActive() - ActionCable.log("Attemped to open WebSocket, but existing socket is #{@getState()}") + ActionCable.log("Attempted to open WebSocket, but existing socket is #{@getState()}") throw new Error("Existing connection must be closed before opening") else ActionCable.log("Opening WebSocket, current state is #{@getState()}") diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb index 714d9887d4..464d0581dd 100644 --- a/actioncable/lib/action_cable/channel/base.rb +++ b/actioncable/lib/action_cable/channel/base.rb @@ -160,7 +160,10 @@ module ActionCable action = extract_action(data) if processable_action?(action) - dispatch_action(action, data) + payload = { channel_class: self.class.name, action: action, data: data } + ActiveSupport::Notifications.instrument("perform_action.action_cable", payload) do + dispatch_action(action, data) + end else logger.error "Unable to process #{action_signature(action, data)}" end @@ -192,7 +195,11 @@ module ActionCable # the proper channel identifier marked as the recipient. def transmit(data, via: nil) logger.info "#{self.class.name} transmitting #{data.inspect.truncate(300)}".tap { |m| m << " (via #{via})" if via } - connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data) + + payload = { channel_class: self.class.name, data: data, via: via } + ActiveSupport::Notifications.instrument("transmit.action_cable", payload) do + connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, message: data) + end end def defer_subscription_confirmation! @@ -265,8 +272,11 @@ module ActionCable def transmit_subscription_confirmation unless subscription_confirmation_sent? logger.info "#{self.class.name} is transmitting the subscription confirmation" - connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:confirmation]) - @subscription_confirmation_sent = true + + ActiveSupport::Notifications.instrument("transmit_subscription_confirmation.action_cable", channel_class: self.class.name) do + connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:confirmation]) + @subscription_confirmation_sent = true + end end end @@ -277,7 +287,10 @@ module ActionCable def transmit_subscription_rejection logger.info "#{self.class.name} is transmitting the subscription rejection" - connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:rejection]) + + ActiveSupport::Notifications.instrument("transmit_subscription_rejection.action_cable", channel_class: self.class.name) do + connection.transmit ActiveSupport::JSON.encode(identifier: @identifier, type: ActionCable::INTERNAL[:message_types][:rejection]) + end end end end diff --git a/actioncable/lib/action_cable/connection/client_socket.rb b/actioncable/lib/action_cable/connection/client_socket.rb index 9e4dbcd6e6..7d6de78582 100644 --- a/actioncable/lib/action_cable/connection/client_socket.rb +++ b/actioncable/lib/action_cable/connection/client_socket.rb @@ -71,6 +71,8 @@ module ActionCable def write(data) @stream.write(data) + rescue => e + emit_error e.message end def transmit(message) diff --git a/actioncable/lib/action_cable/connection/faye_client_socket.rb b/actioncable/lib/action_cable/connection/faye_client_socket.rb index c9139b6858..47d09a9e14 100644 --- a/actioncable/lib/action_cable/connection/faye_client_socket.rb +++ b/actioncable/lib/action_cable/connection/faye_client_socket.rb @@ -36,6 +36,7 @@ module ActionCable @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 diff --git a/actioncable/lib/action_cable/connection/faye_event_loop.rb b/actioncable/lib/action_cable/connection/faye_event_loop.rb index 8b70f3d84e..9c44b38bc3 100644 --- a/actioncable/lib/action_cable/connection/faye_event_loop.rb +++ b/actioncable/lib/action_cable/connection/faye_event_loop.rb @@ -36,7 +36,7 @@ module ActionCable end def shutdown - inner.cancel + @inner.cancel end end end diff --git a/actioncable/lib/action_cable/connection/stream.rb b/actioncable/lib/action_cable/connection/stream.rb index 2d97b28c09..0cf59091bc 100644 --- a/actioncable/lib/action_cable/connection/stream.rb +++ b/actioncable/lib/action_cable/connection/stream.rb @@ -29,7 +29,7 @@ module ActionCable def write(data) return @rack_hijack_io.write(data) if @rack_hijack_io return @stream_send.call(data) if @stream_send - rescue EOFError + rescue EOFError, Errno::ECONNRESET @socket_object.client_gone end diff --git a/actioncable/test/channel/base_test.rb b/actioncable/test/channel/base_test.rb index d41bf3064b..bed54eb6b3 100644 --- a/actioncable/test/channel/base_test.rb +++ b/actioncable/test/channel/base_test.rb @@ -166,6 +166,81 @@ class ActionCable::Channel::BaseTest < ActiveSupport::TestCase end end + test "notification for perform_action" do + begin + events = [] + ActiveSupport::Notifications.subscribe "perform_action.action_cable" do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + data = {'action' => :speak, 'content' => 'hello'} + @channel.perform_action data + + assert_equal 1, events.length + assert_equal 'perform_action.action_cable', events[0].name + assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class] + assert_equal :speak, events[0].payload[:action] + assert_equal data, events[0].payload[:data] + ensure + ActiveSupport::Notifications.unsubscribe "perform_action.action_cable" + end + end + + test "notification for transmit" do + begin + events = [] + ActiveSupport::Notifications.subscribe 'transmit.action_cable' do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + @channel.perform_action 'action' => :get_latest + expected_data = {data: 'latest'} + + assert_equal 1, events.length + assert_equal 'transmit.action_cable', events[0].name + assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class] + assert_equal expected_data, events[0].payload[:data] + assert_nil events[0].payload[:via] + ensure + ActiveSupport::Notifications.unsubscribe 'transmit.action_cable' + end + end + + test "notification for transmit_subscription_confirmation" do + begin + events = [] + ActiveSupport::Notifications.subscribe 'transmit_subscription_confirmation.action_cable' do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + @channel.stubs(:subscription_confirmation_sent?).returns(false) + @channel.send(:transmit_subscription_confirmation) + + assert_equal 1, events.length + assert_equal 'transmit_subscription_confirmation.action_cable', events[0].name + assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class] + ensure + ActiveSupport::Notifications.unsubscribe 'transmit_subscription_confirmation.action_cable' + end + end + + test "notification for transmit_subscription_rejection" do + begin + events = [] + ActiveSupport::Notifications.subscribe 'transmit_subscription_rejection.action_cable' do |*args| + events << ActiveSupport::Notifications::Event.new(*args) + end + + @channel.send(:transmit_subscription_rejection) + + assert_equal 1, events.length + assert_equal 'transmit_subscription_rejection.action_cable', events[0].name + assert_equal 'ActionCable::Channel::BaseTest::ChatChannel', events[0].payload[:channel_class] + ensure + ActiveSupport::Notifications.unsubscribe 'transmit_subscription_rejection.action_cable' + end + end + private def assert_logged(message) old_logger = @connection.logger diff --git a/actioncable/test/connection/client_socket_test.rb b/actioncable/test/connection/client_socket_test.rb new file mode 100644 index 0000000000..dd730e348f --- /dev/null +++ b/actioncable/test/connection/client_socket_test.rb @@ -0,0 +1,65 @@ +require 'test_helper' +require 'stubs/test_server' + +class ActionCable::Connection::StreamTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + attr_reader :websocket, :subscriptions, :message_buffer, :connected + attr_reader :errors + + def initialize(*) + super + @errors = [] + end + + def connect + @connected = true + end + + def disconnect + @connected = false + end + + def send_async(method, *args) + send method, *args + end + + def on_error(message) + @errors << message + end + end + + setup do + @server = TestServer.new + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + end + + test 'delegate socket errors to on_error handler' do + skip if ENV['FAYE'].present? + + run_in_eventmachine do + connection = open_connection + + # Internal hax = :( + client = connection.websocket.send(:websocket) + client.instance_variable_get('@stream').expects(:write).raises('foo') + client.expects(:client_gone).never + + client.write('boo') + assert_equal %w[ foo ], connection.errors + end + end + + private + def open_connection + env = Rack::MockRequest.env_for '/test', + 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket', + 'HTTP_HOST' => 'localhost', 'HTTP_ORIGIN' => 'http://rubyonrails.com' + env['rack.hijack'] = -> { env['rack.hijack_io'] = StringIO.new } + + Connection.new(@server, env).tap do |connection| + connection.process + connection.send :handle_open + assert connection.connected + end + end +end diff --git a/actioncable/test/connection/stream_test.rb b/actioncable/test/connection/stream_test.rb new file mode 100644 index 0000000000..d5aad63648 --- /dev/null +++ b/actioncable/test/connection/stream_test.rb @@ -0,0 +1,67 @@ +require 'test_helper' +require 'stubs/test_server' + +class ActionCable::Connection::StreamTest < ActionCable::TestCase + class Connection < ActionCable::Connection::Base + attr_reader :websocket, :subscriptions, :message_buffer, :connected + attr_reader :errors + + def initialize(*) + super + @errors = [] + end + + def connect + @connected = true + end + + def disconnect + @connected = false + end + + def send_async(method, *args) + send method, *args + end + + def on_error(message) + @errors << message + end + end + + setup do + @server = TestServer.new + @server.config.allowed_request_origins = %w( http://rubyonrails.com ) + end + + [ 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 + + # Internal hax = :( + client = connection.websocket.send(:websocket) + client.instance_variable_get('@stream').instance_variable_get('@rack_hijack_io').expects(:write).raises(closed_exception, 'foo') + client.expects(:client_gone) + + client.write('boo') + assert_equal [], connection.errors + end + end + end + + private + def open_connection + env = Rack::MockRequest.env_for '/test', + 'HTTP_CONNECTION' => 'upgrade', 'HTTP_UPGRADE' => 'websocket', + 'HTTP_HOST' => 'localhost', 'HTTP_ORIGIN' => 'http://rubyonrails.com' + env['rack.hijack'] = -> { env['rack.hijack_io'] = StringIO.new } + + Connection.new(@server, env).tap do |connection| + connection.process + connection.send :handle_open + assert connection.connected + end + end +end diff --git a/actioncable/test/subscription_adapter/evented_redis_test.rb b/actioncable/test/subscription_adapter/evented_redis_test.rb index 70333e51bd..6d20e6ed78 100644 --- a/actioncable/test/subscription_adapter/evented_redis_test.rb +++ b/actioncable/test/subscription_adapter/evented_redis_test.rb @@ -4,6 +4,17 @@ require_relative './common' class EventedRedisAdapterTest < ActionCable::TestCase include CommonSubscriptionAdapterTest + def setup + super + + # em-hiredis is warning-rich + @previous_verbose, $VERBOSE = $VERBOSE, nil + end + + def teardown + $VERBOSE = @previous_verbose + end + def cable_config { adapter: 'evented_redis', url: 'redis://127.0.0.1:6379/12' } end diff --git a/actioncable/test/subscription_adapter/postgresql_test.rb b/actioncable/test/subscription_adapter/postgresql_test.rb index 3b4fadbb2a..214352a0b2 100644 --- a/actioncable/test/subscription_adapter/postgresql_test.rb +++ b/actioncable/test/subscription_adapter/postgresql_test.rb @@ -21,6 +21,7 @@ class PostgresqlAdapterTest < ActionCable::TestCase begin ActiveRecord::Base.connection rescue + @rx_adapter = @tx_adapter = nil skip "Couldn't connect to PostgreSQL: #{database_config.inspect}" end |