diff options
| -rw-r--r-- | actioncable/lib/action_cable/subscription_adapter/postgresql.rb | 32 | ||||
| -rw-r--r-- | actioncable/test/subscription_adapter/postgresql_test.rb | 23 |
2 files changed, 47 insertions, 8 deletions
diff --git a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb index bdab5205ec..e7be4c606e 100644 --- a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb +++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb @@ -11,7 +11,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)}, '#{pg_conn.escape_string(payload)}'") end end @@ -28,14 +28,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 @@ -45,6 +55,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() @@ -60,7 +76,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/test/subscription_adapter/postgresql_test.rb b/actioncable/test/subscription_adapter/postgresql_test.rb index beb6efab28..7c2608a7b8 100644 --- a/actioncable/test/subscription_adapter/postgresql_test.rb +++ b/actioncable/test/subscription_adapter/postgresql_test.rb @@ -37,4 +37,27 @@ class PostgresqlAdapterTest < ActionCable::TestCase def cable_config { adapter: "postgresql" } end + + def test_clear_active_record_connections_adapter_still_works + server = ActionCable::Server::Base.new + server.config.cable = cable_config.with_indifferent_access + server.config.logger = Logger.new(StringIO.new).tap { |l| l.level = Logger::UNKNOWN } + + adapter_klass = Class.new(server.config.pubsub_adapter) do + def active? + !@listener.nil? + end + end + + adapter = adapter_klass.new(server) + + subscribe_as_queue("channel", adapter) do |queue| + adapter.broadcast("channel", "hello world") + assert_equal "hello world", queue.pop + end + + ActiveRecord::Base.clear_reloadable_connections! + + assert adapter.active? + end end |