aboutsummaryrefslogtreecommitdiffstats
path: root/actioncable/lib/action_cable
diff options
context:
space:
mode:
Diffstat (limited to 'actioncable/lib/action_cable')
-rw-r--r--actioncable/lib/action_cable/channel.rb1
-rw-r--r--actioncable/lib/action_cable/channel/base.rb6
-rw-r--r--actioncable/lib/action_cable/channel/test_case.rb271
-rw-r--r--actioncable/lib/action_cable/connection/authorization.rb2
-rw-r--r--actioncable/lib/action_cable/connection/base.rb11
-rw-r--r--actioncable/lib/action_cable/connection/stream.rb6
-rw-r--r--actioncable/lib/action_cable/server/base.rb4
-rw-r--r--actioncable/lib/action_cable/subscription_adapter.rb1
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/postgresql.rb32
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/redis.rb3
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/test.rb40
-rw-r--r--actioncable/lib/action_cable/test_case.rb11
-rw-r--r--actioncable/lib/action_cable/test_helper.rb133
13 files changed, 504 insertions, 17 deletions
diff --git a/actioncable/lib/action_cable/channel.rb b/actioncable/lib/action_cable/channel.rb
index d2f6fbbbc7..d5118b9dc9 100644
--- a/actioncable/lib/action_cable/channel.rb
+++ b/actioncable/lib/action_cable/channel.rb
@@ -11,6 +11,7 @@ module ActionCable
autoload :Naming
autoload :PeriodicTimers
autoload :Streams
+ autoload :TestCase
end
end
end
diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb
index c5ad749bfe..ad0d3685cd 100644
--- a/actioncable/lib/action_cable/channel/base.rb
+++ b/actioncable/lib/action_cable/channel/base.rb
@@ -1,6 +1,7 @@
# frozen_string_literal: true
require "set"
+require "active_support/rescuable"
module ActionCable
module Channel
@@ -99,6 +100,7 @@ module ActionCable
include Streams
include Naming
include Broadcasting
+ include ActiveSupport::Rescuable
attr_reader :params, :connection, :identifier
delegate :logger, to: :connection
@@ -267,10 +269,12 @@ module ActionCable
else
public_send action
end
+ rescue Exception => exception
+ rescue_with_handler(exception) || raise
end
def action_signature(action, data)
- "#{self.class.name}##{action}".dup.tap do |signature|
+ (+"#{self.class.name}##{action}").tap do |signature|
if (arguments = data.except("action")).any?
signature << "(#{arguments.inspect})"
end
diff --git a/actioncable/lib/action_cable/channel/test_case.rb b/actioncable/lib/action_cable/channel/test_case.rb
new file mode 100644
index 0000000000..dd2762ccd0
--- /dev/null
+++ b/actioncable/lib/action_cable/channel/test_case.rb
@@ -0,0 +1,271 @@
+# frozen_string_literal: true
+
+require "active_support"
+require "active_support/test_case"
+require "active_support/core_ext/hash/indifferent_access"
+require "json"
+
+module ActionCable
+ module Channel
+ class NonInferrableChannelError < ::StandardError
+ def initialize(name)
+ super "Unable to determine the channel to test from #{name}. " +
+ "You'll need to specify it using `tests YourChannel` in your " +
+ "test case definition."
+ end
+ end
+
+ # Stub `stream_from` to track streams for the channel.
+ # Add public aliases for `subscription_confirmation_sent?` and
+ # `subscription_rejected?`.
+ module ChannelStub
+ def confirmed?
+ subscription_confirmation_sent?
+ end
+
+ def rejected?
+ subscription_rejected?
+ end
+
+ def stream_from(broadcasting, *)
+ streams << broadcasting
+ end
+
+ def stop_all_streams
+ @_streams = []
+ end
+
+ def streams
+ @_streams ||= []
+ end
+
+ # Make periodic timers no-op
+ def start_periodic_timers; end
+ alias stop_periodic_timers start_periodic_timers
+ end
+
+ class ConnectionStub
+ attr_reader :transmissions, :identifiers, :subscriptions, :logger
+
+ def initialize(identifiers = {})
+ @transmissions = []
+
+ identifiers.each do |identifier, val|
+ define_singleton_method(identifier) { val }
+ end
+
+ @subscriptions = ActionCable::Connection::Subscriptions.new(self)
+ @identifiers = identifiers.keys
+ @logger = ActiveSupport::TaggedLogging.new ActiveSupport::Logger.new(StringIO.new)
+ end
+
+ def transmit(cable_message)
+ transmissions << cable_message.with_indifferent_access
+ end
+ end
+
+ # Superclass for Action Cable channel functional tests.
+ #
+ # == Basic example
+ #
+ # Functional tests are written as follows:
+ # 1. First, one uses the +subscribe+ method to simulate subscription creation.
+ # 2. Then, one asserts whether the current state is as expected. "State" can be anything:
+ # transmitted messages, subscribed streams, etc.
+ #
+ # For example:
+ #
+ # class ChatChannelTest < ActionCable::Channel::TestCase
+ # def test_subscribed_with_room_number
+ # # Simulate a subscription creation
+ # subscribe room_number: 1
+ #
+ # # Asserts that the subscription was successfully created
+ # assert subscription.confirmed?
+ #
+ # # Asserts that the channel subscribes connection to a stream
+ # assert_equal "chat_1", streams.last
+ # end
+ #
+ # def test_does_not_subscribe_without_room_number
+ # subscribe
+ #
+ # # Asserts that the subscription was rejected
+ # assert subscription.rejected?
+ # end
+ # end
+ #
+ # You can also perform actions:
+ # def test_perform_speak
+ # subscribe room_number: 1
+ #
+ # perform :speak, message: "Hello, Rails!"
+ #
+ # assert_equal "Hello, Rails!", transmissions.last["text"]
+ # end
+ #
+ # == Special methods
+ #
+ # ActionCable::Channel::TestCase will also automatically provide the following instance
+ # methods for use in the tests:
+ #
+ # <b>connection</b>::
+ # An ActionCable::Channel::ConnectionStub, representing the current HTTP connection.
+ # <b>subscription</b>::
+ # An instance of the current channel, created when you call `subscribe`.
+ # <b>transmissions</b>::
+ # A list of all messages that have been transmitted into the channel.
+ # <b>streams</b>::
+ # A list of all created streams subscriptions (as identifiers) for the subscription.
+ #
+ #
+ # == Channel is automatically inferred
+ #
+ # ActionCable::Channel::TestCase will automatically infer the channel under test
+ # from the test class name. If the channel cannot be inferred from the test
+ # class name, you can explicitly set it with +tests+.
+ #
+ # class SpecialEdgeCaseChannelTest < ActionCable::Channel::TestCase
+ # tests SpecialChannel
+ # end
+ #
+ # == Specifying connection identifiers
+ #
+ # You need to set up your connection manually to provide values for the identifiers.
+ # To do this just use:
+ #
+ # stub_connection(user: users[:john])
+ #
+ # == Testing broadcasting
+ #
+ # ActionCable::Channel::TestCase enhances ActionCable::TestHelper assertions (e.g.
+ # +assert_broadcasts+) to handle broadcasting to models:
+ #
+ #
+ # # in your channel
+ # def speak(data)
+ # broadcast_to room, text: data["message"]
+ # end
+ #
+ # def test_speak
+ # subscribe room_id: rooms[:chat].id
+ #
+ # assert_broadcasts_on(rooms[:chat], text: "Hello, Rails!") do
+ # perform :speak, message: "Hello, Rails!"
+ # end
+ # end
+ class TestCase < ActiveSupport::TestCase
+ module Behavior
+ extend ActiveSupport::Concern
+
+ include ActiveSupport::Testing::ConstantLookup
+ include ActionCable::TestHelper
+
+ CHANNEL_IDENTIFIER = "test_stub"
+
+ included do
+ class_attribute :_channel_class
+
+ attr_reader :connection, :subscription
+ delegate :streams, to: :subscription
+
+ ActiveSupport.run_load_hooks(:action_cable_channel_test_case, self)
+ end
+
+ module ClassMethods
+ def tests(channel)
+ case channel
+ when String, Symbol
+ self._channel_class = channel.to_s.camelize.constantize
+ when Module
+ self._channel_class = channel
+ else
+ raise NonInferrableChannelError.new(channel)
+ end
+ end
+
+ def channel_class
+ if channel = self._channel_class
+ channel
+ else
+ tests determine_default_channel(name)
+ end
+ end
+
+ def determine_default_channel(name)
+ channel = determine_constant_from_test_name(name) do |constant|
+ Class === constant && constant < ActionCable::Channel::Base
+ end
+ raise NonInferrableChannelError.new(name) if channel.nil?
+ channel
+ end
+ end
+
+ # Setup test connection with the specified identifiers:
+ #
+ # class ApplicationCable < ActionCable::Connection::Base
+ # identified_by :user, :token
+ # end
+ #
+ # stub_connection(user: users[:john], token: 'my-secret-token')
+ def stub_connection(identifiers = {})
+ @connection = ConnectionStub.new(identifiers)
+ end
+
+ # Subscribe to the channel under test. Optionally pass subscription parameters as a Hash.
+ def subscribe(params = {})
+ @connection ||= stub_connection
+ @subscription = self.class.channel_class.new(connection, CHANNEL_IDENTIFIER, params.with_indifferent_access)
+ @subscription.singleton_class.include(ChannelStub)
+ @subscription.subscribe_to_channel
+ @subscription
+ end
+
+ # Unsubscribe the subscription under test.
+ def unsubscribe
+ check_subscribed!
+ subscription.unsubscribe_from_channel
+ end
+
+ # Perform action on a channel.
+ #
+ # NOTE: Must be subscribed.
+ def perform(action, data = {})
+ check_subscribed!
+ subscription.perform_action(data.stringify_keys.merge("action" => action.to_s))
+ end
+
+ # Returns messages transmitted into channel
+ def transmissions
+ # Return only directly sent message (via #transmit)
+ connection.transmissions.map { |data| data["message"] }.compact
+ end
+
+ # Enhance TestHelper assertions to handle non-String
+ # broadcastings
+ def assert_broadcasts(stream_or_object, *args)
+ super(broadcasting_for(stream_or_object), *args)
+ end
+
+ def assert_broadcast_on(stream_or_object, *args)
+ super(broadcasting_for(stream_or_object), *args)
+ end
+
+ private
+ def check_subscribed!
+ raise "Must be subscribed!" if subscription.nil? || subscription.rejected?
+ end
+
+ def broadcasting_for(stream_or_object)
+ return stream_or_object if stream_or_object.is_a?(String)
+
+ self.class.channel_class.broadcasting_for(
+ [self.class.channel_class.channel_name, stream_or_object]
+ )
+ end
+ end
+
+ include Behavior
+ end
+ end
+end
diff --git a/actioncable/lib/action_cable/connection/authorization.rb b/actioncable/lib/action_cable/connection/authorization.rb
index a22179d988..aef3386f71 100644
--- a/actioncable/lib/action_cable/connection/authorization.rb
+++ b/actioncable/lib/action_cable/connection/authorization.rb
@@ -5,7 +5,7 @@ module ActionCable
module Authorization
class UnauthorizedError < StandardError; end
- # Closes the \WebSocket connection if it is open and returns a 404 "File not Found" response.
+ # Closes the WebSocket connection if it is open and returns a 404 "File not Found" response.
def reject_unauthorized_connection
logger.error "An unauthorized connection attempt was rejected"
raise UnauthorizedError
diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb
index 11a1f1a5e8..0044afad98 100644
--- a/actioncable/lib/action_cable/connection/base.rb
+++ b/actioncable/lib/action_cable/connection/base.rb
@@ -95,7 +95,12 @@ module ActionCable
end
# Close the WebSocket connection.
- def close
+ def close(reason: nil, reconnect: true)
+ transmit(
+ type: ActionCable::INTERNAL[:message_types][:disconnect],
+ reason: reason,
+ reconnect: reconnect
+ )
websocket.close
end
@@ -170,7 +175,7 @@ module ActionCable
message_buffer.process!
server.add_connection(self)
rescue ActionCable::Connection::Authorization::UnauthorizedError
- respond_to_invalid_request
+ close(reason: ActionCable::INTERNAL[:disconnect_reasons][:unauthorized], reconnect: false) if websocket.alive?
end
def handle_close
@@ -211,7 +216,7 @@ module ActionCable
end
def respond_to_invalid_request
- close if websocket.alive?
+ close(reason: ActionCable::INTERNAL[:disconnect_reasons][:invalid_request]) if websocket.alive?
logger.error invalid_request_message
logger.info finished_request_message
diff --git a/actioncable/lib/action_cable/connection/stream.rb b/actioncable/lib/action_cable/connection/stream.rb
index 4873026b71..e658948a55 100644
--- a/actioncable/lib/action_cable/connection/stream.rb
+++ b/actioncable/lib/action_cable/connection/stream.rb
@@ -98,8 +98,10 @@ module ActionCable
def hijack_rack_socket
return unless @socket_object.env["rack.hijack"]
- @socket_object.env["rack.hijack"].call
- @rack_hijack_io = @socket_object.env["rack.hijack_io"]
+ # This should return the underlying io according to the SPEC:
+ @rack_hijack_io = @socket_object.env["rack.hijack"].call
+ # Retain existing behaviour if required:
+ @rack_hijack_io ||= @socket_object.env["rack.hijack_io"]
@event_loop.attach(@rack_hijack_io, self)
end
diff --git a/actioncable/lib/action_cable/server/base.rb b/actioncable/lib/action_cable/server/base.rb
index 1ee03f6dfc..2b9e1cba3b 100644
--- a/actioncable/lib/action_cable/server/base.rb
+++ b/actioncable/lib/action_cable/server/base.rb
@@ -36,7 +36,9 @@ module ActionCable
end
def restart
- connections.each(&:close)
+ connections.each do |connection|
+ connection.close(reason: ActionCable::INTERNAL[:disconnect_reasons][:server_restart])
+ end
@mutex.synchronize do
# Shutdown the worker pool
diff --git a/actioncable/lib/action_cable/subscription_adapter.rb b/actioncable/lib/action_cable/subscription_adapter.rb
index bcece8d33b..6a9d5c2080 100644
--- a/actioncable/lib/action_cable/subscription_adapter.rb
+++ b/actioncable/lib/action_cable/subscription_adapter.rb
@@ -5,6 +5,7 @@ module ActionCable
extend ActiveSupport::Autoload
autoload :Base
+ autoload :Test
autoload :SubscriberMap
autoload :ChannelPrefix
end
diff --git a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
index e384ea4afb..50ec438c3a 100644
--- a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
@@ -14,7 +14,7 @@ module ActionCable
end
def broadcast(channel, payload)
- with_connection do |pg_conn|
+ with_broadcast_connection do |pg_conn|
pg_conn.exec("NOTIFY #{pg_conn.escape_identifier(channel_identifier(channel))}, '#{pg_conn.escape_string(payload)}'")
end
end
@@ -31,14 +31,24 @@ module ActionCable
listener.shutdown
end
- def with_connection(&block) # :nodoc:
- ActiveRecord::Base.connection_pool.with_connection do |ar_conn|
- pg_conn = ar_conn.raw_connection
+ def with_subscriptions_connection(&block) # :nodoc:
+ ar_conn = ActiveRecord::Base.connection_pool.checkout.tap do |conn|
+ # Action Cable is taking ownership over this database connection, and
+ # will perform the necessary cleanup tasks
+ ActiveRecord::Base.connection_pool.remove(conn)
+ end
+ pg_conn = ar_conn.raw_connection
- unless pg_conn.is_a?(PG::Connection)
- raise "The Active Record database must be PostgreSQL in order to use the PostgreSQL Action Cable storage adapter"
- end
+ verify!(pg_conn)
+ yield pg_conn
+ ensure
+ ar_conn.disconnect!
+ end
+ def with_broadcast_connection(&block) # :nodoc:
+ ActiveRecord::Base.connection_pool.with_connection do |ar_conn|
+ pg_conn = ar_conn.raw_connection
+ verify!(pg_conn)
yield pg_conn
end
end
@@ -52,6 +62,12 @@ module ActionCable
@listener || @server.mutex.synchronize { @listener ||= Listener.new(self, @server.event_loop) }
end
+ def verify!(pg_conn)
+ unless pg_conn.is_a?(PG::Connection)
+ raise "The Active Record database must be PostgreSQL in order to use the PostgreSQL Action Cable storage adapter"
+ end
+ end
+
class Listener < SubscriberMap
def initialize(adapter, event_loop)
super()
@@ -67,7 +83,7 @@ module ActionCable
end
def listen
- @adapter.with_connection do |pg_conn|
+ @adapter.with_subscriptions_connection do |pg_conn|
catch :shutdown do
loop do
until @queue.empty?
diff --git a/actioncable/lib/action_cable/subscription_adapter/redis.rb b/actioncable/lib/action_cable/subscription_adapter/redis.rb
index c28951608f..ad8fa52760 100644
--- a/actioncable/lib/action_cable/subscription_adapter/redis.rb
+++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb
@@ -13,7 +13,8 @@ module ActionCable
# Overwrite this factory method for Redis connections if you want to use a different Redis library than the redis gem.
# This is needed, for example, when using Makara proxies for distributed Redis.
cattr_accessor :redis_connector, default: ->(config) do
- ::Redis.new(config.slice(:url, :host, :port, :db, :password))
+ config[:id] ||= "ActionCable-PID-#{$$}"
+ ::Redis.new(config.slice(:url, :host, :port, :db, :password, :id))
end
def initialize(*)
diff --git a/actioncable/lib/action_cable/subscription_adapter/test.rb b/actioncable/lib/action_cable/subscription_adapter/test.rb
new file mode 100644
index 0000000000..ce604cc88e
--- /dev/null
+++ b/actioncable/lib/action_cable/subscription_adapter/test.rb
@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+require_relative "async"
+
+module ActionCable
+ module SubscriptionAdapter
+ # == Test adapter for Action Cable
+ #
+ # The test adapter should be used only in testing. Along with
+ # <tt>ActionCable::TestHelper</tt> it makes a great tool to test your Rails application.
+ #
+ # To use the test adapter set +adapter+ value to +test+ in your +config/cable.yml+ file.
+ #
+ # NOTE: Test adapter extends the <tt>ActionCable::SubscriptionsAdapter::Async</tt> adapter,
+ # so it could be used in system tests too.
+ class Test < Async
+ def broadcast(channel, payload)
+ broadcasts(channel) << payload
+ super
+ end
+
+ def broadcasts(channel)
+ channels_data[channel] ||= []
+ end
+
+ def clear_messages(channel)
+ channels_data[channel] = []
+ end
+
+ def clear
+ @channels_data = nil
+ end
+
+ private
+ def channels_data
+ @channels_data ||= {}
+ end
+ end
+ end
+end
diff --git a/actioncable/lib/action_cable/test_case.rb b/actioncable/lib/action_cable/test_case.rb
new file mode 100644
index 0000000000..d153259bf6
--- /dev/null
+++ b/actioncable/lib/action_cable/test_case.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+require "active_support/test_case"
+
+module ActionCable
+ class TestCase < ActiveSupport::TestCase
+ include ActionCable::TestHelper
+
+ ActiveSupport.run_load_hooks(:action_cable_test_case, self)
+ end
+end
diff --git a/actioncable/lib/action_cable/test_helper.rb b/actioncable/lib/action_cable/test_helper.rb
new file mode 100644
index 0000000000..7bc877663c
--- /dev/null
+++ b/actioncable/lib/action_cable/test_helper.rb
@@ -0,0 +1,133 @@
+# frozen_string_literal: true
+
+module ActionCable
+ # Provides helper methods for testing Action Cable broadcasting
+ module TestHelper
+ def before_setup # :nodoc:
+ server = ActionCable.server
+ test_adapter = ActionCable::SubscriptionAdapter::Test.new(server)
+
+ @old_pubsub_adapter = server.pubsub
+
+ server.instance_variable_set(:@pubsub, test_adapter)
+ super
+ end
+
+ def after_teardown # :nodoc:
+ super
+ ActionCable.server.instance_variable_set(:@pubsub, @old_pubsub_adapter)
+ end
+
+ # Asserts that the number of broadcasted messages to the stream matches the given number.
+ #
+ # def test_broadcasts
+ # assert_broadcasts 'messages', 0
+ # ActionCable.server.broadcast 'messages', { text: 'hello' }
+ # assert_broadcasts 'messages', 1
+ # ActionCable.server.broadcast 'messages', { text: 'world' }
+ # assert_broadcasts 'messages', 2
+ # end
+ #
+ # If a block is passed, that block should cause the specified number of
+ # messages to be broadcasted.
+ #
+ # def test_broadcasts_again
+ # assert_broadcasts('messages', 1) do
+ # ActionCable.server.broadcast 'messages', { text: 'hello' }
+ # end
+ #
+ # assert_broadcasts('messages', 2) do
+ # ActionCable.server.broadcast 'messages', { text: 'hi' }
+ # ActionCable.server.broadcast 'messages', { text: 'how are you?' }
+ # end
+ # end
+ #
+ def assert_broadcasts(stream, number)
+ if block_given?
+ original_count = broadcasts_size(stream)
+ yield
+ new_count = broadcasts_size(stream)
+ actual_count = new_count - original_count
+ else
+ actual_count = broadcasts_size(stream)
+ end
+
+ assert_equal number, actual_count, "#{number} broadcasts to #{stream} expected, but #{actual_count} were sent"
+ end
+
+ # Asserts that no messages have been sent to the stream.
+ #
+ # def test_no_broadcasts
+ # assert_no_broadcasts 'messages'
+ # ActionCable.server.broadcast 'messages', { text: 'hi' }
+ # assert_broadcasts 'messages', 1
+ # end
+ #
+ # If a block is passed, that block should not cause any message to be sent.
+ #
+ # def test_broadcasts_again
+ # assert_no_broadcasts 'messages' do
+ # # No job messages should be sent from this block
+ # end
+ # end
+ #
+ # Note: This assertion is simply a shortcut for:
+ #
+ # assert_broadcasts 'messages', 0, &block
+ #
+ def assert_no_broadcasts(stream, &block)
+ assert_broadcasts stream, 0, &block
+ end
+
+ # Asserts that the specified message has been sent to the stream.
+ #
+ # def test_assert_transmited_message
+ # ActionCable.server.broadcast 'messages', text: 'hello'
+ # assert_broadcast_on('messages', text: 'hello')
+ # end
+ #
+ # If a block is passed, that block should cause a message with the specified data to be sent.
+ #
+ # def test_assert_broadcast_on_again
+ # assert_broadcast_on('messages', text: 'hello') do
+ # ActionCable.server.broadcast 'messages', text: 'hello'
+ # end
+ # end
+ #
+ def assert_broadcast_on(stream, data)
+ # Encode to JSON and back–we want to use this value to compare
+ # with decoded JSON.
+ # Comparing JSON strings doesn't work due to the order if the keys.
+ serialized_msg =
+ ActiveSupport::JSON.decode(ActiveSupport::JSON.encode(data))
+
+ new_messages = broadcasts(stream)
+ if block_given?
+ old_messages = new_messages
+ clear_messages(stream)
+
+ yield
+ new_messages = broadcasts(stream)
+ clear_messages(stream)
+
+ # Restore all sent messages
+ (old_messages + new_messages).each { |m| pubsub_adapter.broadcast(stream, m) }
+ end
+
+ message = new_messages.find { |msg| ActiveSupport::JSON.decode(msg) == serialized_msg }
+
+ assert message, "No messages sent with #{data} to #{stream}"
+ end
+
+ def pubsub_adapter # :nodoc:
+ ActionCable.server.pubsub
+ end
+
+ delegate :broadcasts, :clear_messages, to: :pubsub_adapter
+
+ private
+ def broadcasts_size(channel)
+ broadcasts(channel).size
+ end
+ end
+end