stream.rb in async-http-0.47.0

- old
+ new

@@ -24,10 +24,11 @@
 module Async
 	module HTTP
 		module Protocol
 			module HTTP2
 				class Stream < ::Protocol::HTTP2::Stream
+					# A writable body which requests window updates when data is read from it.
 					class Input < Body::Writable
 						def initialize(stream, length)
 							super(length)
 							
 							@stream = stream
@@ -111,11 +112,13 @@
 						private
 						
 						def stream(task)
 							task.annotate("Streaming #{@body} to #{@stream}.")
 							
-							@body.call(Body::Stream.new(@stream.input, self))
+							input = @stream.wait_for_input
+							
+							@body.call(Body::Stream.new(input, self))
 						rescue Async::Stop
 							# Ignore.
 						end
 						
 						# Reads chunks from the given body and writes them to the stream as fast as possible.
@@ -210,10 +213,16 @@
 						Async.logger.error(self, error)
 						
 						send_reset_stream(error.code)
 					end
 					
+					def wait_for_input
+						return @input
+					end
+					
+					# Prepare the input stream which will be used for incoming data frames.
+					# @return [Input] the input body.
 					def prepare_input(length)
 						if @input.nil?
 							@input = Input.new(self, length)
 						else
 							raise ArgumentError, "Input body already prepared!"
@@ -221,10 +230,10 @@
 					end
 					
 					def update_local_window(frame)
 						consume_local_window(frame)
 						
-						# This is done on demand.
+						# This is done on demand in `Input#read`:
 						# request_window_update
 					end
 					
 					def process_data(frame)
 						data = frame.unpack