Close persistent proxy clients in dependency order

lib/async/http/faraday/clients.rb

@@ -86,7 +86,7 @@ def close
 					clients = @clients.values
 					@clients.clear
 
-					clients.each(&:close)
+					clients.reverse_each(&:close)
 				end
 
 				# Lookup or create a client for the given endpoint.

test/async/http/faraday/clients.rb

@@ -6,6 +6,72 @@
 require "async/http/faraday/clients"
 require "async/http/middleware/location_redirector"
 
+require "sus/fixtures/async/http/server_context"
+
+require "socket"
+
+class ConnectProxy
+	def initialize
+		@server = TCPServer.new("127.0.0.1", 0)
+		@workers = []
+		@thread = Thread.new do
+			loop do
+				client = @server.accept
+				@workers << Thread.new{tunnel(client)}
+			end
+		rescue IOError, Errno::EBADF
+			# The server was closed.
+		end
+	end
+
+	def endpoint
+		Async::HTTP::Endpoint.parse("http://127.0.0.1:#{@server.local_address.ip_port}")
+	end
+
+	def close
+		@server.close
+		@thread.join
+		@workers.each(&:join)
+	end
+
+	private
+
+	def tunnel(client)
+		request = client.gets
+		return unless request&.start_with?("CONNECT ")
+
+		authority = request.split(" ", 3)[1]
+		host, port = authority.split(":", 2)
+
+		while (line = client.gets) && line != "\r\n"
+		end
+
+		upstream = TCPSocket.new(host, port)
+		client.write("HTTP/1.1 200 Connection established\r\n\r\n")
+
+		copy(client, upstream)
+	ensure
+		client&.close
+		upstream&.close
+	end
+
+	def copy(client, upstream)
+		pump = lambda do |input, output|
+			Thread.new do
+				IO.copy_stream(input, output)
+			rescue IOError, SystemCallError
+				# Either side of the tunnel was closed.
+			ensure
+				output.close_write rescue nil
+			end
+		end
+
+		threads = [pump.call(client, upstream), pump.call(upstream, client)]
+
+		threads.each(&:join)
+	end
+end
+
 describe Async::HTTP::Faraday::PersistentClients do
 	let(:clients) {subject.new}
 
@@ -66,6 +132,31 @@
 
 			clients.close
 		end
+
+		with "a CONNECT proxy" do
+			include Sus::Fixtures::Async::HTTP::ServerContext
+
+			it "closes the tunnel before the proxy client" do
+				proxy = ConnectProxy.new
+				endpoint = Async::HTTP::Endpoint.parse(bound_url)
+				closed = false
+
+				clients.with_proxied_client(proxy.endpoint, endpoint) do |client|
+					response = client.get("/")
+					expect(response.read).to be == "Hello World!"
+				end
+
+				cached_clients = clients.instance_variable_get(:@clients).values
+
+				Async::Task.current.with_timeout(1) do
+					clients.close
+					closed = true
+				end
+			ensure
+				cached_clients&.reverse_each(&:close) unless closed
+				proxy&.close
+			end
+		end
 	end
 end