Better HTTP/1 connection handling.

Author: ioquatix

async-http.gemspec

@@ -29,7 +29,7 @@ Gem::Specification.new do |spec|
 	spec.add_dependency "io-endpoint", "~> 0.11"
 	spec.add_dependency "io-stream", "~> 0.4"
 	spec.add_dependency "protocol-http", "~> 0.37"
-	spec.add_dependency "protocol-http1", "~> 0.25"
-	spec.add_dependency "protocol-http2", "~> 0.18"
+	spec.add_dependency "protocol-http1", "~> 0.27"
+	spec.add_dependency "protocol-http2", "~> 0.19"
 	spec.add_dependency "traces", ">= 0.10"
 end

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

@@ -18,11 +18,12 @@ def initialize(...)
 
 					attr_accessor :pool
 
-					def closed!
+					def closed(error = nil)
 						super
 
 						if pool = @pool
 							@pool = nil
+							# If the connection is not reusable, this will retire it from the connection pool and invoke `#close`.
 							pool.release(self)
 						end
 					end
@@ -50,30 +51,32 @@ def call(request, task: Task.current)
 								task.async(annotation: "Upgrading request...") do
 									# If this fails, this connection will be closed.
 									write_upgrade_body(protocol, body)
+								rescue => error
+									self.close(error)
 								end
 							elsif request.connect?
 								task.async(annotation: "Tunnneling request...") do
 									write_tunnel_body(@version, body)
+								rescue => error
+									self.close(error)
 								end
 							else
 								task.async(annotation: "Streaming request...") do
 									# Once we start writing the body, we can't recover if the request fails. That's because the body might be generated dynamically, streaming, etc.
 									write_body(@version, body, false, trailer)
+								rescue => error
+									self.close(error)
 								end
 							end
 						elsif protocol = request.protocol
 							write_upgrade_body(protocol)
 						else
-							write_body(@version, body, false, trailer)
+							write_body(@version, request.body, false, trailer)
 						end
 
-						response = Response.read(self, request)
-						
-						return response
-					rescue
-						# This will ensure that #reusable? returns false.
-						self.close
-						
+						return Response.read(self, request)
+					rescue => error
+						self.close(error)
 						raise
 					end
 				end

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

@@ -43,6 +43,8 @@ def http2?
 
 					def read_line?
 						@stream.read_until(CRLF)
+					rescue Errno::ECONNRESET
+						return nil
 					end
 
 					def read_line

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

@@ -32,12 +32,12 @@ def read
 					end
 
 					def close(error = nil)
+						super
+						
 						unless @closed.resolved?
 							@error = error
 							@closed.value = true
 						end
-						
-						super
 					end
 
 					def wait(persistent = true)

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

@@ -22,7 +22,7 @@ def initialize(...)
 						@ready = Async::Notification.new
 					end
 
-					def closed!
+					def closed(error = nil)
 						super
 
 						@ready.signal
@@ -38,14 +38,12 @@ def fail_request(status)
 					end
 
 					def next_request
-						# Wait for the connection to become idle before reading the next request:
-						unless idle?
+						if closed?
+							return nil
+						elsif !idle?
 							@ready.wait
 						end
 
-						# The default is true.
-						return unless @persistent
-						
 						# Read an incoming request:
 						return unless request = Request.read(self)
 
@@ -90,37 +88,41 @@ def each(task: Task.current)
 										# We force a 101 response if the protocol is upgraded - HTTP/2 CONNECT will return 200 for success, but this won't be understood by HTTP/1 clients:
 										write_response(@version, 101, response.headers)
 
-										stream = write_upgrade_body(protocol)
-										
 										# At this point, the request body is hijacked, so we don't want to call #finish below.
 										request = nil
 										response = nil
 
-										# In the case of streaming, `finishable` should wrap a `Remainder` body, which we can safely discard later on.
-										body.call(stream)
+										if body.stream?
+											return body.call(write_upgrade_body(protocol))
+										else
+											write_upgrade_body(protocol, body)
+										end
 									elsif response.status == 101
 										# This code path is to support legacy behavior where the response status is set to 101, but the protocol is not upgraded. This may not be a valid use case, but it is supported for compatibility. We expect the response headers to contain the `upgrade` header.
 										write_response(@version, response.status, response.headers)
 
-										stream = write_tunnel_body(version)
-										
 										# Same as above:
 										request = nil
 										response = nil
 
-										body&.call(stream)
+										if body.stream?
+											return body.call(write_tunnel_body(version))
+										else
+											write_tunnel_body(version, body)
+										end
 									else
 										write_response(@version, response.status, response.headers)
 
 										if request.connect? and response.success?
-											stream = write_tunnel_body(version)
-											
 											# Same as above:
 											request = nil
 											response = nil
 
-											# We must return here as no further request processing can be done:
-											return body.call(stream)
+											if body.stream?
+												return body.call(write_tunnel_body(version))
+											else
+												write_tunnel_body(version, body)
+											end
 										else
 											head = request.head?
 

lib/async/http/protocol/http2/stream.rb

@@ -62,9 +62,9 @@ def process_headers(frame)
 						end
 
 						# TODO this might need to be in an ensure block:
-						if @input and frame.end_stream?
-							@input.close_write
+						if input = @input and frame.end_stream?
 							@input = nil
+							input.close_write
 						end
 					rescue ::Protocol::HTTP2::HeaderError => error
 						Console.logger.debug(self, error)
@@ -123,6 +123,8 @@ def send_body(body, trailer = nil)
 
 					# Called when the output terminates normally.
 					def finish_output(error = nil)
+						return if self.closed?
+						
 						trailer = @output&.trailer
 
 						@output = nil
@@ -152,14 +154,14 @@ def window_updated(size)
 					def closed(error)
 						super
 
-						if @input
-							@input.close_write(error)
+						if input = @input
 							@input = nil
+							input.close_write(error)
 						end
 
-						if @output
-							@output.stop(error)
+						if output = @output
 							@output = nil
+							output.stop(error)
 						end
 
 						if pool = @pool and @connection

test/async/http/proxy.rb

@@ -153,8 +153,6 @@
 							upstream.write(chunk)
 							upstream.flush
 						end
-					rescue Async::Wrapper::Cancelled
-						#ignore
 					ensure
 						Console.logger.debug(self) {"Finished writing to upstream..."}
 						upstream.close_write