Add support for client/server CONNECT proxy.

Author: ioquatix

async-http.gemspec

@@ -17,10 +17,10 @@ Gem::Specification.new do |spec|
 	spec.require_paths = ["lib"]
 
 	spec.add_dependency("async", "~> 1.19")
-	spec.add_dependency("async-io", "~> 1.24")
+	spec.add_dependency("async-io", "~> 1.25")
 
 	spec.add_dependency("protocol-http", "~> 0.12.0")
-	spec.add_dependency("protocol-http1", "~> 0.8.0")
+	spec.add_dependency("protocol-http1", "~> 0.9.0")
 	spec.add_dependency("protocol-http2", "~> 0.9.0")
 
 	# spec.add_dependency("openssl")

lib/async/http/body/pipe.rb

@@ -0,0 +1,100 @@
+# frozen_string_literal: true
+#
+# Copyright, 2019, 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_relative 'writable'
+
+require 'forwardable'
+
+module Async
+	module HTTP
+		module Body
+			class Pipe
+				extend Forwardable
+				
+				def initialize(input, output = Writable.new, task: Task.current)
+					@input = input
+					@output = output
+					
+					head, tail = IO::Socket.pair(Socket::AF_UNIX, Socket::SOCK_STREAM)
+					
+					@head = Async::IO::Stream.new(head)
+					@tail = tail
+					
+					@reader = nil
+					@writer = nil
+					
+					task.async(&self.method(:reader))
+					task.async(&self.method(:writer))
+				end
+				
+				def to_io
+					@tail
+				end
+				
+				def close
+					@reader&.stop
+					@writer&.stop
+					
+					@tail.close
+				end
+				
+				private
+				
+				# Read from the @input stream and write to the head of the pipe.
+				def reader(task)
+					@reader = task
+					
+					task.annotate "pipe reader"
+					
+					while chunk = @input.read
+						@head.write(chunk)
+						@head.flush
+					end
+					
+					@head.close_write
+				ensure
+					@reader = nil
+					@input.close($!)
+					
+					@head.close if @writer.nil?
+				end
+				
+				# Read from the head of the pipe and write to the @output stream.
+				# If the @tail is closed, this will cause chunk to be nil, which in turn will call `@output.close` and `@head.close`
+				def writer(task)
+					@writer = task
+					
+					task.annotate "pipe writer"
+					
+					while chunk = @head.read_partial
+						@output.write(chunk)
+					end
+				ensure
+					@writer = nil
+					@output.close($!)
+					
+					@head.close if @reader.nil?
+				end
+			end
+		end
+	end
+end

lib/async/http/body/stream.rb

@@ -74,10 +74,31 @@ def read(length = nil, buffer = nil)
 					return buffer
 				end
 
+				# Read at most `size` bytes from the stream. Will avoid reading from the underlying stream if possible.
+				def read_partial(size = nil)
+					if @buffer
+						buffer = @buffer
+						@buffer = nil
+					else
+						buffer = read_next
+					end
+					
+					if buffer and size
+						if buffer.bytesize > size
+							@buffer = buffer.byteslice(size, buffer.bytesize)
+							buffer = buffer.byteslice(0, size)
+						end
+					end
+					
+					return buffer
+				end
+				
 				def read_nonblock(length, buffer = nil)
 					@buffer ||= read_next
 					chunk = nil
 
+					return nil if @buffer.nil?
+					
 					if @buffer.bytesize > length
 						chunk = @buffer.byteslice(0, length)
 						@buffer = @buffer.byteslice(length, @buffer.bytesize)

lib/async/http/endpoint.rb

@@ -29,10 +29,10 @@ module Async
 	module HTTP
 		# Represents a way to connect to a remote HTTP server.
 		class Endpoint < Async::IO::Endpoint
-			def self.parse(string, **options)
+			def self.parse(string, endpoint = nil, **options)
 				url = URI.parse(string).normalize
 
-				return self.new(url, nil, **options)
+				return self.new(url, endpoint, **options)
 			end
 
 			# @option scheme [String] the scheme to use, overrides the URL scheme.
@@ -46,7 +46,8 @@ def initialize(url, endpoint = nil, **options)
 				raise ArgumentError, "URL must be absolute (include scheme, host): #{url}" unless url.absolute?
 
 				@url = url
-				@endpoint = endpoint
+				
+				@endpoint = self.build_endpoint(endpoint)
 			end
 
 			def to_url

lib/async/http/pool.rb

@@ -43,13 +43,14 @@ def initialize(limit = nil, &block)
 				@available = Async::Notification.new
 
 				@limit = limit
-				@active = 0
 
 				@constructor = block
 			end
 
 			# The number of allocated resources.
-			attr :active
+			def active
+				@resources.count
+			end
 
 			# Whether there are resources which are currently in use.
 			def busy?
@@ -60,6 +61,11 @@ def busy?
 				return false
 			end
 
+			# Wait until a pool resource has been freed.
+			def wait
+				@available.wait
+			end
+			
 			# All allocated resources.
 			attr :resources
 
@@ -92,8 +98,6 @@ def release(resource)
 			def close
 				@resources.each_key(&:close)
 				@resources.clear
-				
-				@active = 0
 			end
 
 			def to_s
@@ -121,8 +125,6 @@ def retire(resource)
 
 				@resources.delete(resource)
 
-				@active -= 1
-				
 				resource.close
 
 				@available.signal
@@ -163,11 +165,9 @@ def available_resource
 					end
 				end
 
-				if !@limit or @active < @limit
+				if !@limit or self.active < @limit
 					Async.logger.debug(self) {"No resources resources, allocating new one..."}
 
-					@active += 1
-					
 					return create
 				end
 

lib/async/http/protocol/http1/client.rb

@@ -48,6 +48,12 @@ def call(request, task: Task.current)
 									# If this fails, this connection will be closed.
 									write_upgrade_body(protocol, body)
 								end
+							elsif request.connect?
+								task.async do |subtask|
+									subtask.annotate("Tunnelling body.")
+									
+									write_tunnel_body(@version, body)
+								end
 							else
 								task.async do |subtask|
 									subtask.annotate("Streaming body.")

lib/async/http/protocol/http1/server.rb

@@ -37,7 +37,7 @@ def next_request
 						# Read an incoming request:
 						return unless request = Request.read(self)
 
-						unless persistent?(request.version, request.headers)
+						unless persistent?(request.version, request.method, request.headers)
 							@persistent = false
 						end
 
@@ -58,14 +58,6 @@ def each(task: Task.current)
 							return if @stream.nil? or @stream.closed?
 
 							if response
-								# Try to avoid holding on to request, to minimse GC overhead:
-								head = request.head?
-								
-								unless request.body?
-									# If there is no body, #finish is a no-op.
-									request = nil
-								end
-								
 								write_response(@version, response.status, response.headers)
 
 								body = response.body
@@ -79,8 +71,21 @@ def each(task: Task.current)
 									# We also don't want to hold on to the response object:
 									response = nil
 
+									body.call(stream)
+								elsif body and request.connect?
+									stream = write_tunnel_body(request.version)
+									
+									# Same as above:
+									request = nil
+									response = nil
+									
 									body.call(stream)
 								else
+									head = request.head?
+									
+									request = nil unless body
+									response = nil
+									
 									write_body(@version, body, head)
 								end
 							else

lib/async/http/proxy.rb

@@ -0,0 +1,70 @@
+# Copyright, 2019, 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_relative 'client'
+require_relative 'endpoint'
+
+require_relative 'body/pipe'
+
+module Async
+	module HTTP
+		class Proxy
+			def self.tcp(client, host, port, headers = [])
+				self.new(client, "#{host}:#{port}", headers)
+			end
+			
+			def initialize(client, address, headers = [])
+				@client = client
+				@address = address
+				@headers = headers
+			end
+			
+			attr :client
+			
+			def close
+				while @client.pool.busy?
+					@client.pool.wait
+				end
+				
+				@client.close
+			end
+			
+			def connect(&block)
+				input = Body::Writable.new
+				
+				response = @client.connect(@address.to_s, @headers, input)
+				
+				pipe = Body::Pipe.new(response.body, input)
+				
+				return pipe.to_io unless block_given?
+				
+				begin
+					yield pipe.to_io
+				ensure
+					pipe.close
+				end
+			end
+			
+			def endpoint(url, **options)
+				Endpoint.parse(url, self, **options)
+			end
+		end
+	end
+end