Revert "Tidy up body code. (#181)"

This reverts commit d5f0b31.

Author: ioquatix

examples/upload/client.rb

@@ -9,27 +9,10 @@
 
 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)
@@ -38,7 +21,7 @@ def read
 		['accept', 'text/plain'],
 	]
 
-	body = Delayed.new(Protocol::HTTP::Body::File.open(File.join(__dir__, "data.txt"), block_size: 32))
+	body = Async::HTTP::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_write
+						body.close
 					end
 
 					response = client.post("/", headers, body)
@@ -187,7 +187,7 @@ module HTTP
 							response_received.wait
 							headers.add('etag', 'abcd')
 
-							body.close_write
+							body.close
 						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_write
+							stream.write content
+							stream.write content
+							stream.close
 						end
 					end
 				end

fixtures/async/http/body/a_writable_body.rb

@@ -0,0 +1,105 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2019-2023, by Samuel Williams.
+
+require 'protocol/http/body/deflate'
+
+module Async
+	module HTTP
+		module Body
+			AWritableBody = Sus::Shared("a writable body") do
+				it "can write and read data" do
+					3.times do |i|
+						body.write("Hello World #{i}")
+						expect(body.read).to be == "Hello World #{i}"
+					end
+				end
+				
+				it "can buffer data in order" do
+					3.times do |i|
+						body.write("Hello World #{i}")
+					end
+					
+					3.times do |i|
+						expect(body.read).to be == "Hello World #{i}"
+					end
+				end
+				
+				with '#join' do
+					it "can join chunks" do
+						3.times do |i|
+							body.write("#{i}")
+						end
+						
+						body.close
+						
+						expect(body.join).to be == "012"
+					end
+				end
+				
+				with '#each' do
+					it "can read all data in order" do
+						3.times do |i|
+							body.write("Hello World #{i}")
+						end
+						
+						body.close
+						
+						3.times do |i|
+							chunk = body.read
+							expect(chunk).to be == "Hello World #{i}"
+						end
+					end
+					
+					it "can propagate failures" do
+						reactor.async do
+							expect do
+								body.each do |chunk|
+									raise RuntimeError.new("It was too big!")
+								end
+							end.to raise_exception(RuntimeError, message: be =~ /big/)
+						end
+						
+						expect{
+							body.write("Beep boop") # This will cause a failure.
+							::Async::Task.current.yield
+							body.write("Beep boop") # This will fail.
+						}.to raise_exception(RuntimeError, message: be =~ /big/)
+					end
+					
+					it "can propagate failures in nested bodies" do
+						nested = ::Protocol::HTTP::Body::Deflate.for(body)
+						
+						reactor.async do
+							expect do
+								nested.each do |chunk|
+									raise RuntimeError.new("It was too big!")
+								end
+							end.to raise_exception(RuntimeError, message: be =~ /big/)
+						end
+						
+						expect{
+							body.write("Beep boop") # This will cause a failure.
+							::Async::Task.current.yield
+							body.write("Beep boop") # This will fail.
+						}.to raise_exception(RuntimeError, message: be =~ /big/)
+					end
+					
+					it "can consume chunks" do
+						body.write("Hello World!")
+						body.close
+						
+						expect(body).not.to be(:empty?)
+						
+						body.each do |chunk|
+							expect(chunk).to be == "Hello World!"
+						end
+						
+						expect(body).to be(:empty?)
+					end
+				end
+			end
+		end
+	end
+end

gems.rb

@@ -20,8 +20,6 @@
 # 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/delayed.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2018-2023, by Samuel Williams.
+# Copyright, 2020, by Bruno Sutic.
+# Copyright, 2023, by Thomas Morgan.
+
+require 'protocol/http/body/wrapper'
+
+module Async
+	module HTTP
+		module Body
+			class Delayed < ::Protocol::HTTP::Body::Wrapper
+				def initialize(body, delay = 0.01)
+					super(body)
+					
+					@delay = delay
+				end
+				
+				def ready?
+					false
+				end
+				
+				def read
+					Async::Task.current.sleep(@delay)
+					
+					return super
+				end
+			end
+		end
+	end
+end

lib/async/http/body/hijack.rb

@@ -36,7 +36,7 @@ def stream?
 				end
 
 				def call(stream)
-					@block.call(stream)
+					return @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(head)
+					@head = ::IO::Stream::Buffered.new(head)
 					@tail = tail
 
 					@reader = nil
@@ -52,10 +52,8 @@ def reader(task)
 					end
 
 					@head.close_write
-				rescue => error
-					raise
 				ensure
-					@input.close(error)
+					@input.close($!)
 
 					close_head if @writer&.finished?
 				end
@@ -70,10 +68,8 @@ def writer(task)
 					while chunk = @head.read_partial
 						@output.write(chunk)
 					end
-				rescue => error
-					raise
 				ensure
-					@output.close_write(error)
+					@output.close($!)
 
 					close_head if @reader&.finished?
 				end

lib/async/http/body/slowloris.rb

@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2019-2023, by Samuel Williams.
+
+require_relative 'writable'
+
+require 'async/clock'
+
+module Async
+	module HTTP
+		module Body
+			# A dynamic body which you can write to and read from.
+			class Slowloris < Writable
+				class ThroughputError < StandardError
+					def initialize(throughput, minimum_throughput, time_since_last_write)
+						super("Slow write: #{throughput.round(1)}bytes/s less than required #{minimum_throughput.round}bytes/s.")
+					end
+				end
+				
+				# In order for this implementation to work correctly, you need to use a LimitedQueue.
+				# @param minimum_throughput [Integer] the minimum bytes per second otherwise this body will be forcefully closed.
+				def initialize(*arguments, minimum_throughput: 1024, **options)
+					super(*arguments, **options)
+					
+					@minimum_throughput = minimum_throughput
+					
+					@last_write_at = nil
+					@last_chunk_size = nil
+				end
+				
+				attr :minimum_throughput
+				
+				# If #read is called regularly to maintain throughput, that is good. If #read is not called, that is a problem. Throughput is dependent on data being available, from #write, so it doesn't seem particularly problimatic to do this check in #write.
+				def write(chunk)
+					if @last_chunk_size
+						time_since_last_write = Async::Clock.now - @last_write_at
+						throughput = @last_chunk_size / time_since_last_write
+						
+						if throughput < @minimum_throughput
+							error = ThroughputError.new(throughput, @minimum_throughput, time_since_last_write)
+							
+							self.close(error)
+						end
+					end
+					
+					super.tap do
+						@last_write_at = Async::Clock.now
+						@last_chunk_size = chunk&.bytesize
+					end
+				end
+			end
+		end
+	end
+end

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

@@ -68,31 +68,19 @@ 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
+										request = nil unless request.body
 										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
+											request = nil unless request.body
 											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 'protocol/http/body/writable'
+require_relative '../../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 < ::Protocol::HTTP::Body::Writable
+				class Input < Body::Writable
 					def initialize(stream, length)
 						super(length)