Reset connections when connection-changing configuration changes

Adds a few basic features around connection and connection manager
management:

* `clear` on connection manager, which calls `finish` on each active
  connection and then disposes of it.

* A centralized cross-thread tracking system for connection managers in
  `StripeClient` and `clear_all_connection_managers` which clears all
  known connection managers across all threads in a thread-safe way.

The addition of these allow us to modify the implementation of some of
our configuration on `Stripe` so that it can reset all currently open
connections when its value changes.

This fixes a currently problem with the library whereby certain
configuration must be set before the first request or it remains fixed
on any open connections. For example, if `Stripe.proxy` is set after a
request is made from the library, it has no effect because the proxy
must have been set when the connection was originally being initialized.

The impetus for getting this out is that I noticed that we will need
this internally in a few places when we're upgrading to stripe-ruby V5.
Those spots used to be able to hack around the unavailability of this
feature by just accessing the Faraday connection directly and resetting
state on it, but in V5 `StripeClient#conn` is gone, and that's no longer
possible.

lib/stripe.rb

@@ -73,9 +73,20 @@ module Stripe
  @enable_telemetry = true
 
  class << self
- attr_accessor :stripe_account, :api_key, :api_base, :verify_ssl_certs,
- :api_version, :client_id, :connect_base, :uploads_base,
- :open_timeout, :read_timeout, :proxy
+ attr_accessor :api_key
+ attr_accessor :api_version
+ attr_accessor :client_id
+ attr_accessor :stripe_account
+
+ # These all get manual attribute writers so that we can reset connections
+ # if they change.
+ attr_reader :api_base
+ attr_reader :connect_base
+ attr_reader :open_timeout
+ attr_reader :proxy
+ attr_reader :read_timeout
+ attr_reader :uploads_base
+ attr_reader :verify_ssl_certs
 
  attr_reader :max_network_retry_delay, :initial_network_retry_delay
  end
@@ -90,6 +101,11 @@ def self.app_info=(info)
  @app_info = info
  end
 
+ def self.api_base=(api_base)
+ @api_base = api_base
+ StripeClient.clear_all_connection_managers
+ end
+
  # The location of a file containing a bundle of CA certificates. By default
  # the library will use an included bundle that can successfully validate
  # Stripe certificates.
@@ -102,6 +118,8 @@ def self.ca_bundle_path=(path)
 
  # empty this field so a new store is initialized
  @ca_store = nil
+
+ StripeClient.clear_all_connection_managers
  end
 
  # A certificate store initialized from the the bundle in #ca_bundle_path and
@@ -121,6 +139,19 @@ def self.ca_store
  end
  end
 
+ def self.connection_base=(connection_base)
+ @connection_base = connection_base
+ StripeClient.clear_all_connection_managers
+ end
+
+ def self.enable_telemetry?
+ @enable_telemetry
+ end
+
+ def self.enable_telemetry=(val)
+ @enable_telemetry = val
+ end
+
  # map to the same values as the standard library's logger
  LEVEL_DEBUG = Logger::DEBUG
  LEVEL_ERROR = Logger::ERROR
@@ -175,12 +206,19 @@ def self.max_network_retries=(val)
  @max_network_retries = val.to_i
  end
 
- def self.enable_telemetry?
- @enable_telemetry
+ def self.open_timeout=(open_timeout)
+ @open_timeout = open_timeout
+ StripeClient.clear_all_connection_managers
  end
 
- def self.enable_telemetry=(val)
- @enable_telemetry = val
+ def self.proxy=(proxy)
+ @proxy = proxy
+ StripeClient.clear_all_connection_managers
+ end
+
+ def self.read_timeout=(read_timeout)
+ @read_timeout = read_timeout
+ StripeClient.clear_all_connection_managers
  end
 
  # Sets some basic information about the running application that's sent along
@@ -196,6 +234,16 @@ def self.set_app_info(name, partner_id: nil, url: nil, version: nil)
  version: version,
  }
  end
+
+ def self.uploads_base=(uploads_base)
+ @uploads_base = uploads_base
+ StripeClient.clear_all_connection_managers
+ end
+
+ def self.verify_ssl_certs=(verify_ssl_certs)
+ @verify_ssl_certs = verify_ssl_certs
+ StripeClient.clear_all_connection_managers
+ end
 end
 
 Stripe.log_level = ENV["STRIPE_LOG"] unless ENV["STRIPE_LOG"].nil?

lib/stripe/connection_manager.rb

@@ -20,6 +20,15 @@ def initialize
  @active_connections = {}
  end
 
+ # Finishes any active connections by closing their TCP connection and
+ # clears them from internal tracking.
+ def clear
+ @active_connections.each do |_, connection|
+ connection.finish
+ end
+ @active_connections = {}
+ end
+
  # Gets a connection for a given URI. This is for internal use only as it's
  # subject to change (we've moved between HTTP client schemes in the past
  # and may do it again).

lib/stripe/stripe_client.rb

@@ -5,6 +5,11 @@ module Stripe
  # recover both a resource a call returns as well as a response object that
  # contains information on the HTTP call.
  class StripeClient
+ # A set of all known connection managers across all threads and a mutex to
+ # synchronize global access to them.
+ @all_connection_managers = []
+ @all_connection_managers_mutex = Mutex.new
+
  attr_accessor :connection_manager
 
  # Initializes a new `StripeClient`. Expects a `ConnectionManager` object,
@@ -24,6 +29,20 @@ def self.active_client
  Thread.current[:stripe_client] || default_client
  end
 
+ # Finishes any active connections by closing their TCP connection and
+ # clears them from internal tracking in all connection managers across all
+ # threads.
+ def self.clear_all_connection_managers
+ # Just a quick path for when configuration is being set for the first
+ # time before any connections have been opened. There is technically some
+ # potential for thread raciness here, but not in a practical sense.
+ return if @all_connection_managers.empty?
+
+ @all_connection_managers_mutex.synchronize do
+ @all_connection_managers.each(&:clear)
+ end
+ end
+
  # A default client for the current thread.
  def self.default_client
  Thread.current[:stripe_client_default_client] ||=
@@ -32,8 +51,15 @@ def self.default_client
 
  # A default connection manager for the current thread.
  def self.default_connection_manager
- Thread.current[:stripe_client_default_connection_manager] ||=
- ConnectionManager.new
+ Thread.current[:stripe_client_default_connection_manager] ||= begin
+ connection_manager = ConnectionManager.new
+
+ @all_connection_managers_mutex.synchronize do
+ @all_connection_managers << connection_manager
+ end
+
+ connection_manager
+ end
  end
 
  # Checks if an error is a problem that we should retry on. This includes

test/stripe/connection_manager_test.rb

@@ -8,6 +8,23 @@ class ConnectionManagerTest < Test::Unit::TestCase
  @manager = Stripe::ConnectionManager.new
  end
 
+ context "#clear" do
+ should "clear any active connections" do
+ stub_request(:post, "#{Stripe.api_base}/path")
+ .to_return(body: JSON.generate(object: "account"))
+
+ # Making a request lets us know that at least one connection is open.
+ @manager.execute_request(:post, "#{Stripe.api_base}/path")
+
+ # Now clear the manager.
+ @manager.clear
+
+ # This check isn't great, but it's otherwise difficult to tell that
+ # anything happened with just the public-facing API.
+ assert_equal({}, @manager.instance_variable_get(:@active_connections))
+ end
+ end
+
  context "#connection_for" do
  should "correctly initialize a connection" do
  old_proxy = Stripe.proxy

test/stripe/stripe_client_test.rb

@@ -17,6 +17,50 @@ class StripeClientTest < Test::Unit::TestCase
  end
  end
 
+ context ".clear_all_connection_managers" do
+ should "clear connection managers across all threads" do
+ stub_request(:post, "#{Stripe.api_base}/path")
+ .to_return(body: JSON.generate(object: "account"))
+
+ num_threads = 3
+
+ # Poorly named class -- note this is actually a concurrent queue.
+ recv_queue = Queue.new
+ send_queue = Queue.new
+
+ threads = num_threads.times.map do |_|
+ Thread.start do
+ # Causes a connection manager to be created on this thread and a
+ # connection within that manager to be created for API access.
+ manager = StripeClient.default_connection_manager
+ manager.execute_request(:post, "#{Stripe.api_base}/path")
+
+ # Signal to the main thread we're ready.
+ recv_queue << true
+
+ # Wait for the main thread to signal continue.
+ send_queue.pop
+
+ # This check isn't great, but it's otherwise difficult to tell that
+ # anything happened with just the public-facing API.
+ assert_equal({}, manager.instance_variable_get(:@active_connections))
+ end
+ end
+
+ # Wait for threads to start up.
+ threads.each { recv_queue.pop }
+
+ # Do the clear (the method we're actually trying to test).
+ StripeClient.clear_all_connection_managers
+
+ # Tell threads to run their check.
+ threads.each { send_queue << true }
+
+ # And finally, give all threads time to perform their check.
+ threads.each(&:join)
+ end
+ end
+
  context ".default_client" do
  should "be a StripeClient" do
  assert_kind_of StripeClient, StripeClient.default_client