Preserve streaming body terminal errors

lib/protocol/http/body/readable.rb

@@ -138,17 +138,24 @@ def stream?
 				# @parameter stream [IO | Object] An `IO`-like object that responds to `#read`, `#write` and `#flush`.
 				# @returns [Boolean] Whether the ownership of the stream was transferred.
 				def call(stream)
-					self.each do |chunk|
-						stream.write(chunk)
-
-						# Flush the stream unless we are immediately expecting more data:
-						unless self.ready?
-							stream.flush
+					begin
+						self.each do |chunk|
+							stream.write(chunk)
+
+							# Flush the stream unless we are immediately expecting more data:
+							unless self.ready?
+								stream.flush
+							end
+						end
+					rescue => error
+						raise
+					ensure
+						if error and stream.respond_to?(:close_with_error)
+							stream.close_with_error(error)
+						else
+							stream.close
 						end
 					end
-				ensure
-					# TODO Should this invoke close_write(error) instead?
-					stream.close
 				end
 
 				# Read all remaining chunks into a buffered body and close the underlying input.

lib/protocol/http/body/reader.rb

@@ -15,33 +15,31 @@ module Reader
 				#
 				# @yields {|chunk| ...} chunks from the body.
 				def each(&block)
-					if @body
-						@body.each(&block)
+					if body = @body
 						@body = nil
+						body.each(&block)
 					end
 				end
 
 				# Reads the entire request/response body.
 				#
 				# @returns [String] the entire body as a string.
 				def read
-					if @body
-						buffer = @body.join
+					if body = @body
 						@body = nil
 
-						return buffer
+						return body.join
 					end
 				end
 
 				# Gracefully finish reading the body. This will buffer the remainder of the body.
 				#
 				# @returns [Buffered] buffers the entire body.
 				def finish
-					if @body
-						body = @body.finish
+					if body = @body
 						@body = nil
 
-						return body
+						return body.finish
 					end
 				end
 
@@ -59,8 +57,9 @@ def discard
 				#
 				# @returns [Reader] itself.
 				def buffered!
-					if @body
-						@body = @body.finish
+					if body = @body
+						@body = nil
+						@body = body.finish
 					end
 
 					# TODO Should this return @body instead? It seems more useful.

lib/protocol/http/body/stream.rb

@@ -375,6 +375,15 @@ def close_write(error = nil)
 					end
 				end
 
+				# Close the stream because an error occurred.
+				#
+				# This explicit operation allows IO-compatible consumers to propagate an error without passing an argument to {close}.
+				#
+				# @parameter error [Exception] The error that caused this stream to be closed.
+				def close_with_error(error)
+					self.close(error)
+				end
+
 				# Close the input and output bodies.
 				#
 				# @parameter error [Exception | Nil] The error that caused this stream to be closed, if any.

lib/protocol/http/body/streamable.rb

@@ -129,7 +129,12 @@ def call(stream)
 						block.call(stream)
 					rescue => error
 						# If, for some reason, the block raises an error, we assume it may not have closed the stream, so we close it here:
-						stream.close
+						if stream.respond_to?(:close_with_error)
+							stream.close_with_error(error)
+						else
+							stream.close
+						end
+
 						raise
 					end
 

lib/protocol/http/body/writable.rb

@@ -21,6 +21,7 @@ class Closed < StandardError
 				def initialize(length = nil, queue: Thread::Queue.new)
 					@length = length
 					@queue = queue
+					@mutex = Thread::Mutex.new
 					@count = 0
 					@error = nil
 				end
@@ -31,19 +32,23 @@ def initialize(length = nil, queue: Thread::Queue.new)
 				# @attribute [Integer] The number of chunks written to the body.
 				attr :count
 
-				# Stop generating output; cause the next call to write to fail with the given error. Does not prevent existing chunks from being read. In other words, this indicates both that no more data will be or should be written to the body.
+				# Stop consuming the body and discard any unread chunks. Future writes will fail with the given error, or {Closed} if no error is given.
 				#
-				# @parameter error [Exception] The error that caused this body to be closed, if any. Will be raised on the next call to {read}.
+				# @parameter error [Exception | Nil] The error that caused this body to be closed, if any.
 				def close(error = nil)
-					@error ||= error
-
-					@queue.clear
-					@queue.close
+					@mutex.synchronize do
+						unless @queue.closed?
+							@error = error
+							@queue.close
+						end
+
+						@queue.clear
+					end
 
 					super
 				end
 
-				# Whether the body is closed. A closed body can not be written to or read from.
+				# Whether the body is closed for writing. Buffered chunks may still be read.
 				#
 				# @returns [Boolean] Whether the body is closed.
 				def closed?
@@ -59,22 +64,17 @@ def ready?
 				#
 				# @returns [Boolean] Whether the body is empty.
 				def empty?
-					@queue.empty? && @queue.closed?
+					@error.nil? && @queue.empty? && @queue.closed?
 				end
 
 				# Read the next available chunk.
 				#
 				# @returns [String | Nil] The next chunk, or `nil` if the body is finished.
 				# @raises [Exception] If the body was closed due to an error.
 				def read
-					if @error
-						raise @error
-					end
-
-					# This operation may result in @error being set.
 					chunk = @queue.pop
 
-					if @error
+					if chunk.nil? and @error
 						raise @error
 					end
 
@@ -87,20 +87,22 @@ def read
 				# @raises [Closed] If the body has been closed without error.
 				# @raises [Exception] If the body has been closed due to an error.
 				def write(chunk)
-					if @queue.closed?
-						raise(@error || Closed)
-					end
-
 					@queue.push(chunk)
 					@count += 1
+				rescue ClosedQueueError
+					raise(@error || Closed)
 				end
 
 				# Signal that no more data will be written to the body.
 				#
 				# @parameter error [Exception] The error that caused this body to be closed, if any.
 				def close_write(error = nil)
-					@error ||= error
-					@queue.close
+					@mutex.synchronize do
+						unless @queue.closed?
+							@error = error
+							@queue.close
+						end
+					end
 				end
 
 				# The output interface for writing chunks to the body.
@@ -127,23 +129,18 @@ def write(chunk)
 
 					# Close the output stream.
 					#
-					# If an error is given, the error will be used to close the body by invoking {close} with the error. Otherwise, only the write side of the body will be closed.
+					# If an error is given, it will be raised by the reader after all buffered chunks have been consumed.
 					#
 					# @parameter error [Exception | Nil] The error that caused this stream to be closed, if any.
 					def close(error = nil)
 						@closed = true
-
-						if error
-							@writable.close(error)
-						else
-							@writable.close_write
-						end
+						@writable.close_write(error)
 					end
 				end
 
 				# Create an output wrapper which can be used to write chunks to the body.
 				#
-				# If a block is given, and the block raises an error, the error will used to close the body by invoking {close} with the error.
+				# If a block is given, and the block raises an error, the reader will receive all buffered chunks followed by that error.
 				#
 				# @yields {|output| ...} if a block is given.
 				# 	@parameter output [Output] The output wrapper.

test/protocol/http/body/readable.rb

@@ -45,6 +45,47 @@
 		end
 	end
 
+	with "#each" do
+		it "passes a read error to close" do
+			error = RuntimeError.new("Could not read the body!")
+			closed_error = nil
+
+			mock(body) do |mock|
+				mock.replace(:read){raise error}
+				mock.replace(:close){|argument = nil| closed_error = argument}
+			end
+
+			raised_error = begin
+				body.each{}
+			rescue => exception
+				exception
+			end
+
+			expect(raised_error).to be_equal(error)
+			expect(closed_error).to be_equal(error)
+		end
+
+		it "passes a consumer error to close" do
+			error = RuntimeError.new("Could not consume the body!")
+			closed_error = nil
+			chunks = ["Hello", nil]
+
+			mock(body) do |mock|
+				mock.replace(:read){chunks.shift}
+				mock.replace(:close){|argument = nil| closed_error = argument}
+			end
+
+			raised_error = begin
+				body.each{raise error}
+			rescue => exception
+				exception
+			end
+
+			expect(raised_error).to be_equal(error)
+			expect(closed_error).to be_equal(error)
+		end
+	end
+
 	with "#call" do
 		let(:output) {Protocol::HTTP::Body::Buffered.new}
 		let(:stream) {Protocol::HTTP::Body::Stream.new(nil, output)}
@@ -72,6 +113,45 @@
 
 			body.call(stream)
 		end
+
+		it "closes a plain IO normally when reading fails" do
+			error = RuntimeError.new("Could not read the body!")
+			stream = StringIO.new
+
+			mock(body) do |mock|
+				mock.replace(:read){raise error}
+			end
+
+			raised_error = begin
+				body.call(stream)
+			rescue => exception
+				exception
+			end
+
+			expect(raised_error).to be_equal(error)
+			expect(stream).to be(:closed?)
+		end
+
+		it "passes a read error to a stream with explicit error closure" do
+			error = RuntimeError.new("Could not read the body!")
+			closed_error = nil
+			stream = Object.new
+			stream.define_singleton_method(:close_with_error){|argument| closed_error = argument}
+
+			mock(body) do |mock|
+				mock.replace(:read){raise error}
+			end
+
+			raised_error = begin
+				body.call(stream)
+			rescue => exception
+				exception
+			end
+
+			expect(raised_error).to be_equal(error)
+			expect(closed_error).to be_equal(error)
+		end
+
 	end
 
 	with "#join" do

test/protocol/http/body/reader.rb

@@ -23,10 +23,35 @@ def initialize(body)
 	let(:body) {Protocol::HTTP::Body::Buffered.wrap("thequickbrownfox")}
 	let(:reader) {TestReader.new(body)}
 
+	with "#each" do
+		it "detaches the body after failure" do
+			mock(body) {|mock| mock.replace(:each){raise "Could not read!"}}
+
+			expect{reader.each{}}.to raise_exception(RuntimeError, message: be =~ /Could not read!/)
+			expect(reader.body).to be_nil
+		end
+	end
+
+	with "#read" do
+		it "detaches the body after failure" do
+			mock(body) {|mock| mock.replace(:join){raise "Could not read!"}}
+
+			expect{reader.read}.to raise_exception(RuntimeError, message: be =~ /Could not read!/)
+			expect(reader.body).to be_nil
+		end
+	end
+
 	with "#finish" do
 		it "returns a buffered representation" do
 			expect(reader.finish).to be == body
 		end
+
+		it "detaches the body after failure" do
+			mock(body) {|mock| mock.replace(:finish){raise "Could not read!"}}
+
+			expect{reader.finish}.to raise_exception(RuntimeError, message: be =~ /Could not read!/)
+			expect(reader.body).to be_nil
+		end
 	end
 
 	with "#discard" do
@@ -41,6 +66,13 @@ def initialize(body)
 			expect(reader.buffered!).to be_equal(reader)
 			expect(reader.body).to be == body
 		end
+
+		it "detaches the body after failure" do
+			mock(body) {|mock| mock.replace(:finish){raise "Could not read!"}}
+
+			expect{reader.buffered!}.to raise_exception(RuntimeError, message: be =~ /Could not read!/)
+			expect(reader.body).to be_nil
+		end
 	end
 
 	with "#close" do