Better handling of state transitions.

lib/async/http/client.rb

@@ -14,7 +14,6 @@
 require "traces/provider"
 
 require_relative "protocol"
-require_relative "body/finishable"
 
 module Async
 	module HTTP

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

@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2024, by Samuel Williams.
+
+require "protocol/http/body/wrapper"
+require "async/variable"
+
+module Async
+	module HTTP
+		module Protocol
+			module HTTP1
+				# Keeps track of whether a body is being read, and if so, waits for it to be closed.
+				class Finishable < ::Protocol::HTTP::Body::Wrapper
+					def initialize(body)
+						super(body)
+
+						@closed = Async::Variable.new
+						@error = nil
+
+						@reading = false
+					end
+
+					def reading?
+						@reading
+					end
+
+					def read
+						@reading = true
+
+						super
+					end
+
+					def close(error = nil)
+						unless @closed.resolved?
+							@error = error
+							@closed.value = true
+						end
+
+						super
+					end
+
+					def wait
+						if @reading
+							@closed.wait
+						else
+							self.discard
+						end
+					end
+
+					def inspect
+						"#<#{self.class} closed=#{@closed} error=#{@error}> | #{super}"
+					end
+				end
+			end
+		end
+	end
+end

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

@@ -7,7 +7,7 @@
 # Copyright, 2024, by Anton Zhuravsky.
 
 require_relative "connection"
-require_relative "../../body/finishable"
+require_relative "finishable"
 
 require "console/event/failure"
 
@@ -16,6 +16,18 @@ module HTTP
 		module Protocol
 			module HTTP1
 				class Server < Connection
+					def initialize(...)
+						super
+
+						@ready = Async::Notification.new
+					end
+
+					def closed!
+						super
+
+						@ready.signal
+					end
+
 					def fail_request(status)
 						@persistent = false
 						write_response(@version, status, {})
@@ -26,6 +38,11 @@ def fail_request(status)
 					end
 
 					def next_request
+						# Wait for the connection to become idle before reading the next request:
+						unless idle?
+							@ready.wait
+						end
+
 						# The default is true.
 						return unless @persistent
 
@@ -49,7 +66,7 @@ def each(task: Task.current)
 
 						while request = next_request
 							if body = request.body
-								finishable = Body::Finishable.new(body)
+								finishable = Finishable.new(body)
 								request.body = finishable
 							end
 
@@ -126,10 +143,8 @@ def each(task: Task.current)
 									request&.finish
 								end
 
+								# Discard or wait for the input body to be consumed:
 								finishable&.wait
-
-								# This ensures we yield at least once every iteration of the loop and allow other fibers to execute.
-								task.yield
 							rescue => error
 								raise
 							ensure