Tidy up body code.

examples/upload/client.rb

@@ -9,10 +9,27 @@
 
 require 'async'
 require 'protocol/http/body/file'
-require 'async/http/body/delayed'
 require 'async/http/client'
 require 'async/http/endpoint'
 
+class Delayed < ::Protocol::HTTP::Body::Wrapper
+	def initialize(body, delay = 0.01)
+		super(body)
+
+		@delay = delay
+	end
+
+	def ready?
+		false
+	end
+
+	def read
+		sleep(@delay)
+
+		return super
+	end
+end
+
 Async do
 	endpoint = Async::HTTP::Endpoint.parse("http://localhost:9222")
 	client = Async::HTTP::Client.new(endpoint, protocol: Async::HTTP::Protocol::HTTP2)
@@ -21,7 +38,7 @@
 		['accept', 'text/plain'],
 	]
 
-	body = Async::HTTP::Body::Delayed.new(Protocol::HTTP::Body::File.open(File.join(__dir__, "data.txt"), block_size: 32))
+	body = Delayed.new(Protocol::HTTP::Body::File.open(File.join(__dir__, "data.txt"), block_size: 32))
 
 	response = client.post(endpoint.path, headers, body)
 

fixtures/async/http/a_protocol.rb

@@ -162,7 +162,7 @@ module HTTP
 						request_received.wait
 						headers.add('etag', 'abcd')
 
-						body.close
+						body.close_write
 					end
 
 					response = client.post("/", headers, body)
@@ -187,7 +187,7 @@ module HTTP
 							response_received.wait
 							headers.add('etag', 'abcd')
 
-							body.close
+							body.close_write
 						end
 
 						::Protocol::HTTP::Response[200, headers, body]
@@ -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

@@ -20,6 +20,8 @@
 # gem "protocol-http2", path: "../protocol-http2"
 # gem "protocol-hpack", path: "../protocol-hpack"
 
+gem "protocol-http", git: "https://github.com/socketry/protocol-http.git"
+
 group :maintenance, optional: true do
 	gem "bake-modernize"
 	gem "bake-gem"

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)