Use Streaming implementation from protocol-http.

Author: ioquatix

gems.rb

@@ -20,8 +20,9 @@
 	gem "decode"
 	gem "rubocop"
 
-	gem "sus-fixtures-async-http", "~> 0.1"
+	gem "sus-fixtures-async"
+	gem "sus-fixtures-async-http"
 
-	gem "bake-test", "~> 0.1"
-	gem "bake-test-external", "~> 0.1"
+	gem "bake-test"
+	gem "bake-test-external"
 end

lib/protocol/rack/body/streaming.rb

@@ -3,83 +3,12 @@
 # Released under the MIT License.
 # Copyright, 2022-2024, by Samuel Williams.
 
-require 'protocol/http/body/readable'
-require 'protocol/http/body/stream'
+require 'protocol/http/body/streamable'
 
 module Protocol
 	module Rack
 		module Body
-			# Wraps a streaming response body into a compatible Protocol::HTTP body.
-			class Streaming < ::Protocol::HTTP::Body::Readable
-				def initialize(block, input = nil)
-					@block = block
-					@input = input
-					@output = nil
-				end
-				
-				attr :block
-				
-				class Output
-					def initialize(input, block)
-						stream = ::Protocol::HTTP::Body::Stream.new(input, self)
-						
-						@from = nil
-						
-						@fiber = Fiber.new do |from|
-							@from = from
-							block.call(stream)
-							@fiber = nil
-						end
-					end
-					
-					def write(chunk)
-						if from = @from
-							@from = nil
-							@from = from.transfer(chunk)
-						else
-							raise RuntimeError, "Stream is not being read!"
-						end
-					end
-					
-					def close(error = nil)
-						@fiber = nil
-						
-						if from = @from
-							@from = nil
-							if error
-								from.raise(error)
-							else
-								from.transfer(nil)
-							end
-						end
-					end
-					
-					def close_write(error = nil)
-						close(error)
-					end
-					
-					def read
-						raise RuntimeError, "Stream is already being read!" if @from
-						
-						@fiber&.transfer(Fiber.current)
-					end
-				end
-				
-				# Invokes the block in a fiber which yields chunks when they are available.
-				def read
-					@output ||= Output.new(@input, @block)
-					return @output.read
-				end
-				
-				def stream?
-					true
-				end
-				
-				def call(stream)
-					raise "Streaming body has already been read!" if @output
-					@block.call(stream)
-				end
-			end
+			Streaming = ::Protocol::HTTP::Body::Streamable::ResponseBody
 		end
 	end
 end

protocol-rack.gemspec

@@ -24,6 +24,6 @@ Gem::Specification.new do |spec|
 
 	spec.required_ruby_version = ">= 3.1"
 
-	spec.add_dependency "protocol-http", "~> 0.27"
+	spec.add_dependency "protocol-http", "~> 0.37"
 	spec.add_dependency "rack", ">= 1.0"
 end

test/protocol/rack/body/streaming.rb

@@ -5,16 +5,14 @@
 
 require 'protocol/rack/body/streaming'
 
+require 'sus/fixtures/async/scheduler_context'
+
 describe Protocol::Rack::Body::Streaming do
+	include Sus::Fixtures::Async::SchedulerContext
+	
 	let(:block) {proc{|stream| stream.write("Hello"); stream.write("World"); stream.close}}
 	let(:body) {subject.new(block)}
 
-	with '#block' do
-		it "should wrap block" do
-			expect(body.block).to be == block
-		end
-	end
-	
 	with '#read' do
 		it "can read the body" do
 			expect(body.read).to be == "Hello"