Rework client handling to enable per-thread client cache.

Author: ioquatix

lib/async/http/faraday/adapter.rb

@@ -16,7 +16,7 @@
 require 'async/http/client'
 require 'async/http/proxy'
 
-require_relative 'client_cache'
+require_relative 'clients'
 
 module Async
 	module HTTP
@@ -71,7 +71,12 @@ def initialize(...)
 					super
 
 					@timeout = @connection_options.delete(:timeout)
-					@clients = ClientCache.new(**@connection_options)
+					
+					if clients = @connection_options.delete(:clients)
+						@clients = clients.call(**@connection_options, &@config_block)
+					else
+						@clients = PersistentClients.new(**@connection_options, &@config_block)
+					end
 				end
 
 				# Close all clients.
@@ -89,19 +94,10 @@ def close
 				def call(env)
 					super
 
-					# for compatibility with the default adapter
+					# For compatibility with the default adapter:
 					env.url.path = '/' if env.url.path.empty?
 
-					Sync do
-						endpoint = Endpoint.new(env.url)
-						
-						if proxy = env.request.proxy
-							proxy_endpoint = Endpoint.new(proxy.uri)
-							client = @clients.proxy_client_for(proxy_endpoint, endpoint)
-						else
-							client = @clients.client_for(endpoint)
-						end
-						
+					with_client(env) do |endpoint, client|
 						if body = env.body
 							# We need to ensure the body is wrapped in a Readable object so that it can be read in chunks:
 							# Faraday's body only responds to `#read`.
@@ -140,6 +136,24 @@ def call(env)
 
 				private
 
+				def with_client(env)
+					Sync do
+						endpoint = Endpoint.new(env.url)
+						
+						if proxy = env.request.proxy
+							proxy_endpoint = Endpoint.new(proxy.uri)
+							
+							@clients.with_proxied_client(proxy_endpoint, endpoint) do |client|
+								yield endpoint, client
+							end
+						else
+							@clients.with_client(endpoint) do |client|
+								yield endpoint, client
+							end
+						end
+					end
+				end
+				
 				def with_timeout(task: Async::Task.current)
 					if @timeout
 						task.with_timeout(@timeout, ::Faraday::TimeoutError) do

lib/async/http/faraday/client_cache.rb

@@ -0,0 +1,173 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2018-2024, by Samuel Williams.
+# Copyright, 2018, by Andreas Garnaes.
+# Copyright, 2019, by Denis Talakevich.
+# Copyright, 2019-2020, by Igor Sidorov.
+# Copyright, 2023, by Genki Takiuchi.
+# Copyright, 2023, by Flavio Fernandes.
+# Copyright, 2024, by Jacob Frautschi.
+
+require 'faraday'
+require 'faraday/adapter'
+require 'kernel/sync'
+
+require 'async/http/client'
+require 'async/http/proxy'
+
+module Async
+	module HTTP
+		module Faraday
+			class Clients
+				def self.call(...)
+					new(...)
+				end
+				
+				def initialize(**options, &block)
+					@options = options
+					@block = block
+				end
+				
+				def close
+				end
+				
+				# Make a new client for the given endpoint.
+				#
+				# @parameter endpoint [IO::Endpoint::Generic] The endpoint to create the client for.
+				def make_client(endpoint)
+					client = Client.new(endpoint, **@options)
+					@block&.call(client)
+					return client
+				end
+				
+				# Get a client for the given endpoint. If a client already exists for the host, it will be reused.
+				#
+				# @parameter endpoint [IO::Endpoint::Generic] The endpoint to get the client for.
+				def with_client(endpoint)
+					client = make_client(endpoint)
+					
+					yield client
+				ensure
+					client&.close
+				end
+				
+				# Get a client for the given proxy endpoint and endpoint. If a client already exists for the host, it will be reused.
+				#
+				# @parameter proxy_endpoint [IO::Endpoint::Generic] The proxy endpoint to use.
+				# @parameter endpoint [IO::Endpoint::Generic] The endpoint to get the client for.
+				def with_proxied_client(proxy_endpoint, endpoint)
+					client = client_for(proxy_endpoint)
+					proxied_client = client.proxied_client(endpoint)
+					
+					yield proxied_client 
+				ensure
+					proxied_client&.close
+					client&.close
+				end
+			end
+			
+			class PersistentClients < Clients
+				def initialize(...)
+					super
+					
+					@clients = {}
+				end
+				
+				def close
+					super
+					
+					clients = @clients.values
+					@clients.clear
+					
+					clients.each(&:close)
+				end
+				
+				# Get the host key for the given endpoint.
+				#
+				# This is used to cache clients for the same host.
+				#
+				# @parameter endpoint [IO::Endpoint::Generic] The endpoint to get the host key for.
+				def host_key(endpoint)
+					url = endpoint.url.dup
+					
+					url.path = ""
+					url.fragment = nil
+					url.query = nil
+					
+					return url
+				end
+				
+				# Get a client for the given endpoint. If a client already exists for the host, it will be reused.
+				#
+				# @parameter endpoint [IO::Endpoint::Generic] The endpoint to get the client for.
+				def client_for(endpoint)
+					key = host_key(endpoint)
+					
+					fetch(key) do
+						make_client
+					end
+				end
+				
+				def with_client(endpoint)
+					yield make_client(endpoint)
+				end
+				
+				# Get a client for the given proxy endpoint and endpoint. If a client already exists for the host, it will be reused.
+				#
+				# @parameter proxy_endpoint [IO::Endpoint::Generic] The proxy endpoint to use.
+				# @parameter endpoint [IO::Endpoint::Generic] The endpoint to get the client for.
+				def with_proxied_client(proxy_endpoint, endpoint)
+					key = [host_key(proxy_endpoint), host_key(endpoint)]
+					
+					proxied_client = fetch(key) do
+						client_for(proxy_endpoint).proxied_client(endpoint)
+					end
+					
+					yield proxied_client
+				end
+				
+				protected
+				
+				def fetch(key)
+					@clients.fetch(key) do
+						@clients[key] = yield
+					end
+				end
+			end
+			
+			class PerThreadPersistentClients
+				def initialize(**options, &block)
+					@options = options
+					@block = block
+					
+					@key = :"#{self.class}_#{object_id}"
+				end
+				
+				def with_client(endpoint, &block)
+					clients.with_client(endpoint, &block)
+				end
+				
+				def with_proxied_client(proxy_endpoint, endpoint, &block)
+					clients.with_proxied_client(proxy_endpoint, endpoint, &block)
+				end
+				
+				def close
+					Thread.list.each do |thread|
+						if clients = thread[@key]
+							clients.close
+							
+							thread[@key] = nil
+						end
+					end
+				end
+				
+				protected
+				
+				def clients
+					Thread.current[@key] ||= PersistentClients.new(**@options, &@block)
+				end
+			end
+		end
+	end
+end

lib/async/http/faraday/clients.rb

@@ -139,6 +139,22 @@ def get_response(url = bound_url, path = '/index', adapter_options: {})
 				expect(response.body).to be == File.read(__FILE__, 128)
 			end
 		end
+		
+		with "a config block" do
+			it "invokes the config block" do
+				config_block_invoked = false
+				
+				adapter = Faraday.new do |builder|
+					builder.adapter :async_http do |client|
+						config_block_invoked = true
+					end
+				end
+				
+				adapter.get(bound_url)
+				
+				expect(config_block_invoked).to be == true
+			end
+		end
 	end
 
 	with "a remote http server" do