Proxy support

.github/workflows/development.yml

@@ -31,6 +31,9 @@ jobs:
           - os: ubuntu
             ruby: head
             experimental: true
+          - os: ubuntu
+            ruby: "2.7"
+            proxy: http://localhost:3128
 
     steps:
     - uses: actions/checkout@v2
@@ -42,3 +45,41 @@ jobs:
     - name: Run tests
       timeout-minutes: 5
       run: ${{matrix.env}} bundle exec rspec
+
+  test-with-proxy:
+    runs-on: ${{matrix.os}}-latest
+    continue-on-error: ${{matrix.experimental}}
+
+    strategy:
+      matrix:
+        os:
+          - ubuntu
+        ruby:
+          - "2.7"
+
+        experimental: [false]
+        env: [""]
+
+        proxy:
+          - http://localhost:3128
+
+    steps:
+    - uses: actions/checkout@v2
+    - uses: ruby/setup-ruby@v1
+      with:
+        ruby-version: ${{matrix.ruby}}
+
+    - name: Install dependencies
+      run: ${{matrix.env}} bundle install
+
+    - name: Prepare squid
+      if: matrix.proxy
+      run: |
+        sudo apt-get install squid
+        sudo systemctl start squid
+
+    - name: Run tests
+      timeout-minutes: 5
+      env:
+        PROXY_URL: ${{matrix.proxy}}
+      run: ${{matrix.env}} bundle exec rspec

lib/async/http/faraday/adapter.rb

@@ -23,7 +23,9 @@
 require 'faraday'
 require 'faraday/adapter'
 require 'kernel/sync'
-require 'async/http/internet'
+
+require 'async/http/client'
+require 'async/http/proxy'
 
 require_relative 'agent'
 
@@ -44,24 +46,87 @@ class Adapter < ::Faraday::Adapter
 					SocketError
 				].freeze
 
-				def initialize(*arguments, **options, &block)
-					super
+				def initialize(*arguments, timeout: nil, **options, &block)
+					super(*arguments, **options)
+
+					@timeout = timeout
+
+					@clients = {}
+					@proxy_clients = {}
+
+					@options = options
+				end
+
+				def make_client(endpoint)
+					Client.new(endpoint, **@connection_options)
+				end
+
+				def host_key(endpoint)
+					url = endpoint.url.dup
 
-					@internet = Async::HTTP::Internet.new
-					@persistent = options.fetch(:persistent, true)
-					@timeout = options[:timeout]
+					url.path = ""
+					url.fragment = nil
+					url.query = nil
+
+					return url
+				end
+
+				def client_for(endpoint)
+					key = host_key(endpoint)
+
+					@clients.fetch(key) do
+						@clients[key] = make_client(endpoint)
+					end
+				end
+
+				def proxy_client_for(proxy_endpoint, endpoint)
+					key = host_key(endpoint)
+
+					@proxy_clients.fetch(key) do
+						client = client_for(proxy_endpoint)
+						@proxy_clients[key] = client.proxied_client(endpoint)
+					end
 				end
 
 				def close
-					@internet.close
+					# The order of operations here is to avoid a race condition between iterating over clients (#close may yield) and creating new clients.
+					proxy_clients = @proxy_clients.values
+					clients = @clients.values
+
+					@proxy_clients.clear
+					@clients.clear
+
+					proxy_clients.each(&:close)
+					clients.each(&:close)
 				end
 
 				def call(env)
 					super
 
 					Sync do
+						endpoint = Endpoint.new(env.url)
+
+						if proxy = env.request.proxy
+							proxy_endpoint = Endpoint.new(proxy.uri)
+							client = self.proxy_client_for(proxy_endpoint, endpoint)
+						else
+							client = self.client_for(endpoint)
+						end
+
+						if body = env.body
+							body = Body::Buffered.wrap(body)
+						end
+
+						if headers = env.request_headers
+							headers = ::Protocol::HTTP::Headers[headers]
+						end
+
+						method = env.method.upcase
+
+						request = ::Protocol::HTTP::Request.new(endpoint.scheme, endpoint.authority, method, endpoint.path, nil, headers, body)
+
 						with_timeout do
-							response = @internet.call(env[:method].to_s.upcase, env[:url].to_s, env[:request_headers], env[:body] || [])
+							response = client.call(request)
 
 							save_response(env, response.status, response.read, response.headers)
 						end

spec/async/http/faraday/adapter/proxy_spec.rb

@@ -0,0 +1,48 @@
+# frozen_string_literal: true
+
+# Copyright, 2017, by Samuel G. D. Williams. <http://www.codeotaku.com>
+# 
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+# 
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+# 
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+require 'async/http/faraday'
+require 'async/http/server'
+require 'async/http/endpoint'
+
+require 'async'
+
+RSpec.describe Async::HTTP::Faraday::Adapter, if: ENV.key?('PROXY_URL') do
+	include_context Async::RSpec::Reactor
+
+	def get_response(url = endpoint.url, path = '/index', adapter_options: {})
+		connection = Faraday.new(url, proxy: ENV['PROXY_URL']) do |faraday|
+			faraday.response :logger
+			faraday.adapter :async_http, **adapter_options
+		end
+
+		connection.get(path)
+	end
+
+	it "can get remote resource via proxy" do
+		Sync do
+			response = get_response('https://www.google.com', '/search?q=cats')
+
+			expect(response).to be_success
+		end
+	end
+end

spec/async/http/faraday/adapter_spec.rb

@@ -34,7 +34,7 @@
 	}
 
 	def run_server(response = Protocol::HTTP::Response[204], response_delay: nil)
-		Async do |task|
+		Sync do |task|
 			begin
 				server_task = task.async do
 					app = Proc.new do
@@ -52,11 +52,11 @@ def run_server(response = Protocol::HTTP::Response[204], response_delay: nil)
 			ensure
 				server_task.stop
 			end
-		end.wait
+		end
 	end
 
 	def get_response(url = endpoint.url, path = '/index', adapter_options: {})
-		connection = Faraday.new(url: url) do |faraday|
+		connection = Faraday.new(url) do |faraday|
 			faraday.response :logger
 			faraday.adapter :async_http, **adapter_options
 		end
@@ -80,7 +80,7 @@ def get_response(url = endpoint.url, path = '/index', adapter_options: {})
 	end
 
 	it "can get remote resource" do
-		Async do
+		Sync do
 			response = get_response('http://www.google.com', '/search?q=cats')
 
 			expect(response).to be_success
@@ -101,14 +101,6 @@ def get_response(url = endpoint.url, path = '/index', adapter_options: {})
 		end
 	end
 
-	it 'closes connection automatically if persistent option is set to false' do
-		run_server do
-			expect do
-				get_response(adapter_options: { persistent: false })
-			end.not_to raise_error
-		end
-	end
-
 	it 'raises an exception if request times out' do
 		delay = 0.1