Improving test coverage.

fixtures/protocol/http/body/a_readable_body.rb

@@ -0,0 +1,14 @@
+module Protocol
+	module HTTP
+		module Body
+			AReadableBody = Sus::Shared("a readable body") do
+				with "#close" do
+					it "should close the body" do
+						body.close
+						expect(body.read).to be_nil
+					end
+				end
+			end
+		end
+	end
+end

lib/protocol/http/body/buffered.rb

@@ -52,8 +52,8 @@ def initialize(chunks = [], length = nil)
 
 				attr :chunks
 
-				def finish
-					self
+				def close(error = nil)
+					@chunks = []
 				end
 
 				def length

lib/protocol/http/body/readable.rb

@@ -17,7 +17,11 @@ module Body
 			#
 			# If you don't want to read from a stream, and instead want to close it immediately, you can call `close` on the body. If the body is already completely consumed, `close` will do nothing, but if there is still data to be read, it will cause the underlying stream to be reset (and possibly closed).
 			class Readable
-				# Close the stream immediately.
+				# Close the stream immediately. After invoking this method, the stream should be considered closed, and all internal resources should be released.
+				#
+				# If an error occured while handling the output, it can be passed as an argument. This may be propagated to the client, for example the client may be informed that the stream was not fully read correctly.
+				#
+				# Invoking `#read` after `#close` will return `nil`.
 				def close(error = nil)
 				end
 

lib/protocol/http/body/streamable.rb

@@ -17,11 +17,14 @@ module Body
 			#
 			# When invoking `call(stream)`, the stream can be read from and written to, and closed. However, the stream is only guaranteed to be open for the duration of the `call(stream)` call. Once the method returns, the stream **should** be closed by the server.
 			module Streamable
+				class ClosedError < StandardError
+				end
+
 				def self.new(*arguments)
 					if arguments.size == 1
-						RequestBody.new(*arguments)
+						DeferredBody.new(*arguments)
 					else
-						ResponseBody.new(*arguments)
+						Body.new(*arguments)
 					end
 				end
 
@@ -39,8 +42,6 @@ def initialize(input, block)
 						@fiber = Fiber.new do |from|
 							@from = from
 							block.call(stream)
-						rescue Closed
-							# Ignore.
 						ensure
 							@fiber = nil
 
@@ -58,16 +59,18 @@ def write(chunk)
 							@from = nil
 							@from = from.transfer(chunk)
 						else
-							raise RuntimeError, "Stream is not being read!"
+							raise ClosedError, "Stream is not being read!"
 						end
 					end
 
 					# Can be invoked by the block to close the stream. Closing the output means that no more chunks will be generated.
 					def close(error = nil)
 						if from = @from
+							# We are closing from within the output fiber, so we need to transfer back to `@from`:
 							@from = nil
 							from.transfer(nil)
 						elsif @fiber
+							# We are closing from outside the output fiber, so we need to resume the fiber appropriately:
 							@from = Fiber.current
 
 							if error
@@ -134,20 +137,21 @@ def call(stream)
 
 					# Closing a stream indicates we are no longer interested in reading from it.
 					def close(error = nil)
-						if output = @output
-							@output = nil
-							output.close(error)
-						end
-
+						$stderr.puts "Closing input: #{@input.inspect}"
 						if input = @input
 							@input = nil
 							input.close(error)
 						end
+
+						if output = @output
+							@output = nil
+							output.close(error)
+						end
 					end
 				end
 
-				# A request body has an extra `stream` method which can be used to stream data into the body, as the response body won't be available until the request has been sent.
-				class RequestBody < Body
+				# A deferred body has an extra `stream` method which can be used to stream data into the body, as the response body won't be available until the request has been sent.
+				class DeferredBody < Body
 					def initialize(block)
 						super(block, Writable.new)
 						@finishing = false
@@ -172,17 +176,6 @@ def stream(input)
 						self.close(error)
 					end
 				end
-
-				def self.request(&block)
-					RequestBody.new(block)
-				end
-
-				class ResponseBody < Body
-				end
-
-				def self.response(request, &block)
-					ResponseBody.new(block, request.body)
-				end
 			end
 		end
 	end

test/protocol/http/body/buffered.rb

@@ -5,11 +5,14 @@
 # Copyright, 2020-2023, by Bruno Sutic.
 
 require 'protocol/http/body/buffered'
+require "protocol/http/body/a_readable_body"
 
 describe Protocol::HTTP::Body::Buffered do
 	let(:source) {["Hello", "World"]}
 	let(:body) {subject.wrap(source)}
 
+	it_behaves_like Protocol::HTTP::Body::AReadableBody
+
 	with ".wrap" do
 		with "an instance of Protocol::HTTP::Body::Readable as a source" do
 			let(:source) {Protocol::HTTP::Body::Readable.new}

test/protocol/http/body/streamable.rb

@@ -62,7 +62,7 @@
 
 				expect do
 					@stream.write("!")
-				end.to raise_exception(RuntimeError, message: be =~ /Stream is not being read!/)
+				end.to raise_exception(Protocol::HTTP::Body::Streamable::ClosedError, message: be =~ /Stream is not being read!/)
 			end
 		end
 	end
@@ -149,6 +149,7 @@
 		let(:block) do
 			proc do |stream|
 				while chunk = stream.read_partial
+					$stderr.puts "Got chunk: #{chunk.inspect}"
 					stream.write(chunk)
 				end
 			end
@@ -170,5 +171,12 @@
 
 			expect(output.string).to be == "Hello World"
 		end
+
+		with '#close' do
+			it "can close the body" do
+				expect(body.read).to be == "Hello"
+				body.close
+			end
+		end
 	end
 end