Use async in tests.

Author: ioquatix

gems.rb

@@ -22,6 +22,8 @@
 	gem "decode"
 	gem "rubocop"
 
+	gem "sus-fixtures-async"
+	
 	gem "bake-test"
 	gem "bake-test-external"
 end

lib/protocol/http/body/streamable.rb

@@ -17,85 +17,6 @@ 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
-				# Raised when an operation is attempted on a closed stream.
-				class ClosedError < StandardError
-				end
-				
-				# Raised when a streaming body is consumed more than once.
-				class ConsumedError < StandardError
-				end
-				
-				# Single value queue that can be used to communicate between fibers.
-				class Queue
-					def self.consumer
-						self.new(Fiber.current, nil)
-					end
-					
-					def self.generator(&block)
-						self.new(Fiber.new(&block), Fiber.current)
-					end
-					
-					def initialize(generator, consumer)
-						@generator = generator
-						@consumer = consumer
-						@closed = false
-					end
-					
-					# The generator fiber can push values into the queue.
-					def push(value)
-						raise ClosedError, "Queue is closed!" if @closed
-						
-						if consumer = @consumer
-							@consumer = nil
-							@generator = Fiber.current
-							
-							consumer.transfer(value)
-						else
-							raise ClosedError, "Queue is not being popped!"
-						end
-					end
-					
-					# The consumer fiber can pop values from the queue.
-					def pop
-						return nil if @closed
-						
-						if generator = @generator
-							@generator = nil
-							@consumer = Fiber.current
-							
-							return generator.transfer
-						else
-							raise ClosedError, "Queue is not being pushed!"
-						end
-					end
-					
-					def close(error = nil)
-						@closed = true
-						
-						if consumer = @consumer
-							@consumer = nil
-							
-							if consumer.alive?
-								@generator = Fiber.current
-								if error
-									consumer.raise(error)
-								else
-									consumer.transfer(nil)
-								end
-							end
-						elsif generator = @generator
-							@generator = nil
-							@consumer = Fiber.current
-							
-							if error
-								generator.raise(error)
-							else
-								generator.transfer(nil)
-							end
-						end
-					end
-				end
-				
 				def self.new(*arguments)
 					if arguments.size == 1
 						DeferredBody.new(*arguments)
@@ -112,73 +33,35 @@ def self.response(request, &block)
 					Body.new(block, request.body)
 				end
 
-				class Input
-					def initialize
-						@queue = Queue.consumer
+				class Output
+					def self.schedule(input, block)
+						self.new(input, block).tap(&:schedule)
 					end
 
-					def read
-						@queue.pop
+					def initialize(input, block)
+						@output = Writable.new
+						@stream = Stream.new(input, @output)
+						@block = block
 					end
 
-					def write(chunk)
-						@queue.push(chunk)
+					def schedule
+						@fiber ||= Fiber.schedule do
+							@block.call(@stream)
+							
+						end
 					end
 
-					def close_write(error = nil)
-						close(error)
+					def read
+						@output.read
 					end
 
 					def close(error = nil)
-						@queue.close(error)
-					end
-					
-					def stream(body)
-						body&.each do |chunk|
-							$stderr.puts "Input stream chunk: #{chunk.inspect}"
-							self.write(chunk)
-						end
-					ensure
-						$stderr.puts "Input stream closed: #{$!}"
-						self.close_write
+						@output.close_write(error)
 					end
 				end
 
-				# Represents an output wrapper around a stream, that can invoke a fiber when `#read`` is called.
-				#
-				# This behaves a little bit like a generator or lazy enumerator, in that it can be used to generate chunks of data on demand.
-				#
-				# When closing the the output, the block is invoked one last time with `nil` to indicate the end of the stream.
-				class Output
-					def initialize(input, block)
-						stream = Stream.new(input, self)
-						
-						@queue = Queue.generator do
-							block.call(stream)
-						end
-					end
-					
-					attr :stream
-					
-					# Generator of the output can write chunks.
-					def write(chunk)
-						@queue.push(chunk)
-					end
-					
-					# Indicates that no further output will be generated.
-					def close_write(error = nil)
-						close(error)
-					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)
-						@queue.close(error)
-					end
-					
-					# Consumer of the output can read chunks.
-					def read
-						@queue.pop
-					end
+				# Raised when a streaming body is consumed more than once.
+				class ConsumedError < StandardError
 				end
 
 				class Body < Readable
@@ -190,6 +73,10 @@ def initialize(block, input = nil)
 
 					attr :block
 
+					attr :input
+					
+					attr :output
+					
 					def stream?
 						true
 					end
@@ -202,7 +89,7 @@ def read
 								raise ConsumedError, "Streaming body has already been consumed!"
 							end
 
-							@output = Output.new(@input, @block)
+							@output = Output.schedule(@input, @block)
 							@block = nil
 						end
 
@@ -231,14 +118,12 @@ def call(stream)
 
 					# Closing a stream indicates we are no longer interested in reading from it.
 					def close(error = nil)
-						$stderr.puts "Closing output #{@output}..."
 						if output = @output
 							@output = nil
 							# Closing the output here may take some time, as it may need to finish handling the stream:
 							output.close(error)
 						end
 
-						$stderr.puts "Closing input #{@input}..."
 						if input = @input
 							@input = nil
 							input.close(error)
@@ -249,29 +134,25 @@ def close(error = nil)
 				# 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, Input.new)
+						super(block, Writable.new)
 					end
 
-					# Closing a stream indicates we are no longer interested in reading from it.
+					# Closing a stream indicates we are no longer interested in reading from it, but in this case that does not mean that the output block is finished generating data.
 					def close(error = nil)
+						if error
+							super
+						end
 					end
 
 					# Stream the response body into the block's input.
 					def stream(body)
-						@input.stream(body)
-						
-						$stderr.puts "Closing output #{@output}..."
-						if output = @output
-							@output = nil
-							# Closing the output here may take some time, as it may need to finish handling the stream:
-							output.close(error)
-						end
-						
-						$stderr.puts "Closing input #{@input}..."
-						if input = @input
-							@input = nil
-							input.close(error)
+						body&.each do |chunk|
+							@input.write(chunk)
 						end
+					rescue => error
+						raise
+					ensure
+						@input.close_write(error)
 					end
 				end
 			end

test/protocol/http/body/streamable.rb

@@ -4,8 +4,11 @@
 # Copyright, 2024, by Samuel Williams.
 
 require 'protocol/http/body/streamable'
+require 'sus/fixtures/async'
 
 describe Protocol::HTTP::Body::Streamable do
+	include Sus::Fixtures::Async::ReactorContext
+	
 	let(:block) do
 		proc do |stream|
 			stream.write("Hello")
@@ -34,37 +37,6 @@
 			expect(body.read).to be == "World"
 			expect(body.read).to be == nil
 		end
-		
-		with "block that doesn't close" do
-			let(:block) do
-				proc do |stream|
-					stream.write("Hello")
-					stream.write("World")
-				end
-			end
-			
-			it "can read the body" do
-				expect(body.read).to be == "Hello"
-				expect(body.read).to be == "World"
-				expect(body.read).to be == nil
-			end
-		end
-		
-		with "a block that allows stream to escape" do
-			let(:block) do
-				proc do |stream|
-					@stream = stream
-				end
-			end
-			
-			it "can read the body" do
-				expect(body.read).to be == nil
-				
-				expect do
-					@stream.write("!")
-				end.to raise_exception(Protocol::HTTP::Body::Streamable::ClosedError, message: be =~ /Stream is not being read!/)
-			end
-		end
 	end
 
 	with '#close_write' do
@@ -146,10 +118,7 @@
 
 		it "can raise an error on the block" do
 			expect(body.read).to be == "Hello"
-			
-			expect do
-				body.close(RuntimeError.new("Oh no!"))
-			end.to raise_exception(RuntimeError, message: be =~ /Oh no!/)
+			body.close(RuntimeError.new("Oh no!"))
 		end
 	end