diff options
Diffstat (limited to 'actioncable/lib')
| -rw-r--r-- | actioncable/lib/action_cable.rb | 2 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/channel/base.rb | 2 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/connection/stream.rb | 6 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/subscription_adapter.rb | 1 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/postgresql.rb | 32 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/redis.rb | 3 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/test.rb | 40 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/test_case.rb | 11 | ||||
| -rw-r--r-- | actioncable/lib/action_cable/test_helper.rb | 133 |
9 files changed, 218 insertions, 12 deletions
diff --git a/actioncable/lib/action_cable.rb b/actioncable/lib/action_cable.rb index e7456e3c1b..35eacc2f4f 100644 --- a/actioncable/lib/action_cable.rb +++ b/actioncable/lib/action_cable.rb @@ -51,4 +51,6 @@ module ActionCable autoload :Channel autoload :RemoteConnections autoload :SubscriptionAdapter + autoload :TestHelper + autoload :TestCase end diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb index c5ad749bfe..70c93ec0f3 100644 --- a/actioncable/lib/action_cable/channel/base.rb +++ b/actioncable/lib/action_cable/channel/base.rb @@ -270,7 +270,7 @@ module ActionCable 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/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/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 |