Improved usage of updated body interface.

Author: ioquatix

fixtures/async/http/a_protocol.rb

@@ -395,9 +395,9 @@ module HTTP
 				let(:app) do
 					::Protocol::HTTP::Middleware.for do |request|
 						Async::HTTP::Body::Hijack.response(request, 200, {}) do |stream|
-							stream.write content
-							stream.write content
-							stream.close
+							stream.write(content)
+							stream.write(content)
+							stream.close_write
 						end
 					end
 				end

gems.rb

@@ -15,7 +15,7 @@
 # gem "traces", path: "../traces"
 # gem "sus-fixtures-async-http", path: "../sus-fixtures-async-http"
 
-# gem "protocol-http", path: "../protocol-http"
+gem "protocol-http", path: "../protocol-http"
 # gem "protocol-http1", path: "../protocol-http1"
 # gem "protocol-http2", path: "../protocol-http2"
 # gem "protocol-hpack", path: "../protocol-hpack"

lib/async/http/body/hijack.rb

@@ -36,7 +36,7 @@ def stream?
 				end
 
 				def call(stream)
-					return @block.call(stream)
+					@block.call(stream)
 				end
 
 				attr :input

lib/async/http/body/pipe.rb

@@ -17,7 +17,7 @@ def initialize(input, output = Writable.new, task: Task.current)
 
 					head, tail = ::Socket.pair(Socket::AF_UNIX, Socket::SOCK_STREAM)
 
-					@head = ::IO::Stream::Buffered.new(head)
+					@head = ::IO::Stream(head)
 					@tail = tail
 
 					@reader = nil
@@ -52,8 +52,10 @@ def reader(task)
 					end
 
 					@head.close_write
+				rescue => error
+					raise
 				ensure
-					@input.close($!)
+					@input.close(error)
 
 					close_head if @writer&.finished?
 				end
@@ -68,8 +70,10 @@ def writer(task)
 					while chunk = @head.read_partial
 						@output.write(chunk)
 					end
+				rescue => error
+					raise
 				ensure
-					@output.close($!)
+					@output.close_write(error)
 
 					close_head if @reader&.finished?
 				end

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

@@ -68,19 +68,31 @@ def each(task: Task.current)
 										stream = write_upgrade_body(protocol)
 
 										# At this point, the request body is hijacked, so we don't want to call #finish below.
-										request = nil unless request.body
+										request = nil
 										response = nil
 
 										# We must return here as no further request processing can be done:
 										return body.call(stream)
+									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(request.version)
+										
+										# Same as above:
+										request = nil
+										response = nil
+										
+										# We must return here as no further request processing can be done:
+										return body&.call(stream)
 									else
 										write_response(@version, response.status, response.headers)
 
 										if request.connect? and response.success?
 											stream = write_tunnel_body(request.version)
 
 											# Same as above:
-											request = nil unless request.body
+											request = nil
 											response = nil
 
 											# We must return here as no further request processing can be done:

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

@@ -3,14 +3,14 @@
 # Released under the MIT License.
 # Copyright, 2020-2023, by Samuel Williams.
 
-require_relative '../../body/writable'
+require 'protocol/http/body/writable'
 
 module Async
 	module HTTP
 		module Protocol
 			module HTTP2
 				# A writable body which requests window updates when data is read from it.
-				class Input < Body::Writable
+				class Input < ::Protocol::HTTP::Body::Writable
 					def initialize(stream, length)
 						super(length)
 

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

@@ -50,18 +50,26 @@ def write(chunk)
 						end
 					end
 
+					def close_write(error = nil)
+						close(error)
+					end
+					
 					# This method should only be called from within the context of the output task.
 					def close(error = nil)
-						if @stream
-							@stream.finish_output(error)
+						stop(error)
+						
+						if stream = @stream
 							@stream = nil
+							stream.finish_output(error)
 						end
 					end
 
 					# This method should only be called from within the context of the HTTP/2 stream.
 					def stop(error)
-						@task&.stop
-						@task = nil
+						if task = @task
+							@task = nil
+							task.stop(error)
+						end
 					end
 
 					private
@@ -70,10 +78,11 @@ def stream(task)
 						task.annotate("Streaming #{@body} to #{@stream}.")
 
 						input = @stream.wait_for_input
+						stream = ::Protocol::HTTP::Body::Stream.new(input, self)
 
-						@body.call(::Protocol::HTTP::Body::Stream.new(input, self))
-					rescue Async::Stop
-						# Ignore.
+						@body.call(stream)
+					rescue => error
+						self.close(error)
 					end
 
 					# Reads chunks from the given body and writes them to the stream as fast as possible.
@@ -86,11 +95,15 @@ def passthrough(task)
 							# chunk.clear unless chunk.frozen?
 							# GC.start
 						end
-						
-						self.close
+					rescue => error
+						raise
 					ensure
-						@body&.close($!)
-						@body = nil
+						if @body
+							@body.close(error)
+							@body = nil
+						end
+						
+						self.close(error)
 					end
 
 					# Send `maximum_size` bytes of data using the specified `stream`. If the buffer has no more chunks, `END_STREAM` will be sent on the final chunk.

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

@@ -98,7 +98,7 @@ def process_data(frame)
 							end
 
 							if frame.end_stream?
-								@input.close
+								@input.close_write
 								@input = nil
 							end
 						end

test/async/http/body.rb

@@ -23,7 +23,7 @@
 						output.write(chunk.reverse)
 					end
 
-					output.close
+					output.close_write
 				end
 
 				Protocol::HTTP::Response[200, [], output]
@@ -35,7 +35,7 @@
 
 			reactor.async do |task|
 				output.write("Hello World!")
-				output.close
+				output.close_write
 			end
 
 			response = client.post("/", {}, output)
@@ -58,7 +58,7 @@
 						notification.wait
 					end
 
-					body.close
+					body.close_write
 				end
 
 				Protocol::HTTP::Response[200, {}, body]

test/async/http/body/hijack.rb

@@ -15,7 +15,7 @@
 			3.times do 
 				stream.write(content)
 			end
-			stream.close
+			stream.close_write
 		end
 	end
 

test/async/http/body/pipe.rb

@@ -20,7 +20,7 @@
 		include Sus::Fixtures::Async::ReactorContext
 
 		let(:input_write_duration) {0}
-		let(:io) { pipe.to_io }
+		let(:io) {pipe.to_io}
 
 		def before
 			super
@@ -31,14 +31,12 @@ def before
 				input.write("#{first} ")
 				sleep(input_write_duration) if input_write_duration > 0
 				input.write(second)
-				input.close
+				input.close_write
 			end
 		end
 
-		def aftrer
+		after do
 			io.close
-			
-			super
 		end
 
 		it "returns an io socket" do

test/async/http/proxy.rb

@@ -57,7 +57,7 @@
 			expect(response).to be(:success?)
 
 			input.write(data)
-			input.close
+			input.close_write
 
 			expect(response.read).to be == data
 		end
@@ -74,7 +74,7 @@
 						stream.flush
 					end
 
-					stream.close
+					stream.close_write
 				end
 			end
 		end