Prefer Protocol::HTTP::Body::Stream.

Author: ioquatix

Diff for: async-http.gemspec

@@ -20,7 +20,7 @@ Gem::Specification.new do |spec|
 	spec.add_dependency "async", ">= 1.25"
 	spec.add_dependency "async-io", ">= 1.28"
 	spec.add_dependency "async-pool", ">= 0.2"
-	spec.add_dependency "protocol-http", "~> 0.22.0"
+	spec.add_dependency "protocol-http", "~> 0.22.9"
 	spec.add_dependency "protocol-http1", "~> 0.14.0"
 	spec.add_dependency "protocol-http2", "~> 0.14.0"
 	spec.add_dependency "traces", "~> 0.4.0"

Diff for: lib/async/http/body/hijack.rb

@@ -21,7 +21,9 @@
 # THE SOFTWARE.
 
 require 'protocol/http/body/readable'
-require_relative 'stream'
+require 'protocol/http/body/stream'
+
+require_relative 'writable'
 
 module Async
 	module HTTP
@@ -42,6 +44,7 @@ def initialize(block, input = nil)
 
 					@task = nil
 					@stream = nil
+					@output = nil
 				end
 
 				# We prefer streaming directly as it's the lowest overhead.
@@ -57,23 +60,18 @@ def call(stream)
 
 				# Has the producer called #finish and has the reader consumed the nil token?
 				def empty?
-					if @stream
-						@stream.empty?
-					else
-						false
-					end
+					@output&.empty?
 				end
 
 				def ready?
-					if @stream
-						@stream.output.ready?
-					end
+					@output&.ready?
 				end
 
 				# Read the next available chunk.
 				def read
-					unless @task
-						@stream = Stream.new(@input)
+					unless @output
+						@output = Writable.new
+						@stream = ::Protocol::HTTP::Body::Stream.new(@input, @output)
 
 						@task = Task.current.async do |task|
 							task.annotate "Streaming hijacked body."
@@ -82,7 +80,7 @@ def read
 						end
 					end
 
-					return @stream.output.read
+					return @output.read
 				end
 
 				def inspect

Diff for: lib/async/http/body/stream.rb

@@ -20,7 +20,7 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 # THE SOFTWARE.
 
-require_relative '../../body/stream'
+require 'protocol/http/body/stream'
 
 module Async
 	module HTTP
@@ -88,7 +88,7 @@ def stream(task)
 
 						input = @stream.wait_for_input
 
-						@body.call(Body::Stream.new(input, self))
+						@body.call(::Protocol::HTTP::Body::Stream.new(input, self))
 					rescue Async::Stop
 						# Ignore.
 					end