WIP

Author: ioquatix

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

@@ -90,37 +90,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/input.rb

@@ -37,6 +37,15 @@ def read
 
 						return chunk
 					end
+					
+					def close(error = nil)
+						super
+						
+						if stream = @stream
+							@stream = nil
+							stream.finish_input(error)
+						end
+					end
 				end
 			end
 		end

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

@@ -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