-
-
Notifications
You must be signed in to change notification settings - Fork 410
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
ConnectionPool adapter #835
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -28,6 +28,7 @@ gem 'mysql2' | |
gem 'pg' | ||
gem 'cuprite' | ||
gem 'puma' | ||
gem 'connection_pool' | ||
|
||
group(:guard) do | ||
gem 'guard' | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,76 @@ | ||
require "connection_pool" | ||
require "observer" | ||
|
||
# Make ConnectionPool observable so we can reset the cache when the pool is reloaded or shutdown | ||
ConnectionPool.class_eval do | ||
module ShutdownObservable | ||
def reload(&block) | ||
super(&block).tap do | ||
changed | ||
notify_observers | ||
end | ||
end | ||
|
||
def shutdown(&block) | ||
super(&block).tap do | ||
changed | ||
notify_observers | ||
end | ||
end | ||
end | ||
|
||
include Observable | ||
prepend ShutdownObservable | ||
end | ||
|
||
# An adapter that uses ConnectionPool to manage connections. | ||
# | ||
# Usage: | ||
# | ||
# # Reuse an existing pool to create new adapters | ||
# pool = ConnectionPool.new(size: 5, timeout: 5) { Redis.new } | ||
# Flipper::Adapters::ConnectionPool.new(pool) do |client| | ||
# Flipper::Adapters::Redis.new(client) | ||
# end | ||
# | ||
# # Create a new pool that returns the adapter | ||
# Flipper::Adapters::ConnectionPool.new(size: 5, timeout: 5) do | ||
# Flipper::Adapters::Redis.new(Redis.new) | ||
# end | ||
class Flipper::Adapters::ConnectionPool | ||
include Flipper::Adapter | ||
|
||
# Use an existing ConnectionPool to initialize and cache new adapter instances. | ||
class Wrapper | ||
def initialize(pool, &initializer) | ||
@pool = pool | ||
@initializer = initializer | ||
@instances = {} | ||
|
||
# Reset the cache when the pool is reloaded or shutdown | ||
@pool.add_observer(self, :reset) | ||
end | ||
|
||
def with(&block) | ||
@pool.with do |resource| | ||
yield @instances[resource] ||= @initializer.call(resource) | ||
end | ||
end | ||
|
||
def reset | ||
@instances.clear | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is threadsafe on MRI due to the GIL, but on other interpreters I could imagine a new object being added to |
||
end | ||
end | ||
|
||
attr_reader :pool | ||
|
||
def initialize(options = {}, &adapter_initializer) | ||
@pool = options.is_a?(ConnectionPool) ? Wrapper.new(options, &adapter_initializer) : ConnectionPool.new(options, &adapter_initializer) | ||
end | ||
|
||
OPERATIONS.each do |method| | ||
define_method(method) do |*args| | ||
@pool.with { |adapter| adapter.public_send(method, *args) } | ||
end | ||
end | ||
end |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,40 @@ | ||
require "connection_pool" | ||
require "flipper/adapters/connection_pool" | ||
require "flipper-redis" | ||
|
||
RSpec.describe Flipper::Adapters::ConnectionPool do | ||
let(:redis_options) { {url: ENV['REDIS_URL']}.compact } | ||
|
||
before do | ||
skip_on_error(Redis::CannotConnectError, 'Redis not available') do | ||
Redis.new(redis_options).flushdb | ||
end | ||
end | ||
|
||
context "with an existing pool" do | ||
let(:pool) do | ||
ConnectionPool.new(size: 1, timeout: 5) { Redis.new(redis_options) } | ||
end | ||
|
||
subject do | ||
described_class.new(pool) { |client| Flipper::Adapters::Redis.new(client) } | ||
end | ||
|
||
it_should_behave_like 'a flipper adapter' | ||
|
||
it "should reset the cache when the pool is reloaded or shutdown" do | ||
subject.get_all | ||
expect { pool.reload { |_| } }.to change { subject.pool.instance_variable_get(:@instances).size }.from(1).to(0) | ||
subject.get_all | ||
expect { pool.shutdown { |_| } }.to change { subject.pool.instance_variable_get(:@instances).size }.from(1).to(0) | ||
end | ||
end | ||
|
||
context "with a new pool" do | ||
subject do | ||
described_class.new(size: 2) { Flipper::Adapters::Redis.new(Redis.new(redis_options)) } | ||
end | ||
|
||
it_should_behave_like 'a flipper adapter' | ||
end | ||
end |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could this hang onto resources for too long in some cases, preventing them from being garbage collected?
ConnectionPool
has areload
capability that will release its hold on the existing resources, but this@instances
hash map will retain them and then grow larger as the pool offers up new resources.I was initially thinking that
ObjectSpace::WeakKeyMap
would solve the problem, but this is a funny situation in which the map keys and values both have references to the same resource.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, good catch. I just pushed 18b2a1b which makes
ConnectionPool
observable. I might open an issue on the uptream repo and see if a change like this would be accepted.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That seems like a good solution. The only other thing that comes to mind is an LRU cache, but I don't know if there's a solid Ruby implementation.