aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/postgresql.rb32
-rw-r--r--actioncable/test/subscription_adapter/postgresql_test.rb23
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