Class: Async::HTTP::Protocol::HTTP2::Stream

Inherits:
Protocol::HTTP2::Stream
  • Object
show all
Defined in:
lib/async/http/protocol/http2/stream.rb

Overview

An HTTP/2 stream that manages headers, input data, and output data for a single request/response exchange.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Stream

Initialize the stream state.



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/async/http/protocol/http2/stream.rb', line 20

def initialize(*)
	super
	
	@headers = nil
	
	@pool = nil
	
	# Input buffer, reading request body, or response body (receive_data):
	@length = nil
	@input = nil
	
	# The application can close its input before the peer finishes sending. HTTP/2 cannot close only the receiving side of a stream, so incoming data is discarded until local output also finishes. At that point, a no-error reset terminates the remaining wire stream.
	@input_closed = false
	
	# Output buffer, writing request body or response body (window_updated):
	@output = nil
end

Instance Attribute Details

#headers ⇒ Object

Returns the value of attribute headers.



38
39
40
 
# File 'lib/async/http/protocol/http2/stream.rb', line 38

def headers
  @headers
end

#input ⇒ Object (readonly)

Returns the value of attribute input.



42
43
44
 
# File 'lib/async/http/protocol/http2/stream.rb', line 42

def input
  @input
end

#pool ⇒ Object

Returns the value of attribute pool.



40
41
42
 
# File 'lib/async/http/protocol/http2/stream.rb', line 40

def pool
  @pool
end

Instance Method Details

#add_header(key, value, trailer: false) ⇒ Object

Add a header to the stream, validating against HTTP/2 constraints.



47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/async/http/protocol/http2/stream.rb', line 47

def add_header(key, value, trailer: false)
	if key == CONNECTION
		raise ::Protocol::HTTP2::HeaderError, "Connection header is not allowed!"
	elsif key.start_with? ":"
		raise ::Protocol::HTTP2::HeaderError, "Invalid pseudo-header #{key}!"
	elsif key =~ /[A-Z]/
		raise ::Protocol::HTTP2::HeaderError, "Invalid upper-case characters in header #{key}!"
	else
		@headers.add(key, value, trailer: trailer)
	end
end

#closed(error) ⇒ Object

When the stream transitions to the closed state, this method is called. There are roughly two ways this can happen:

  • A frame is received which causes this stream to enter the closed state. This method will be invoked from the background reader task.
  • A frame is sent which causes this stream to enter the closed state. This method will be invoked from that task. While the input stream is relatively straight forward, the output stream can trigger the second case above


212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/async/http/protocol/http2/stream.rb', line 212

def closed(error)
	if error.is_a?(::Protocol::HTTP2::StreamError) && error.code == ::Protocol::HTTP2::Error::NO_ERROR
		error = nil
	end
	
	super
	
	if input = @input
		@input = nil
		input.close_write(error)
	end
	
	if output = @output
		@output = nil
		
		if error
			output.stop(error)
		else
			output.close_stream
		end
	end
	
	if pool = @pool and @connection
		pool.release(@connection)
	end
	
	return self
end

#finish_input(input, error = nil) ⇒ Object

Close the application-facing receiving side of the stream. While local output remains active, incoming data is discarded with flow-control updates. Once local output is also closed, the remaining wire stream is terminated without an error.



143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/async/http/protocol/http2/stream.rb', line 143

def finish_input(input, error = nil)
	if @input.equal?(input)
		@input = nil
		@input_closed = true
		
		if error
			send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR)
		else
			close_if_finished
		end
	end
end

#finish_output(error = nil) ⇒ Object

Called when the output terminates normally.



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/async/http/protocol/http2/stream.rb', line 164

def finish_output(error = nil)
	return if self.closed?
	
	trailer = @output&.trailer
	
	@output = nil
	
	if error
		send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR)
	else
		# Write trailer?
		if trailer&.any?
			send_headers(trailer, ::Protocol::HTTP2::END_STREAM)
		else
			send_data(nil, ::Protocol::HTTP2::END_STREAM)
		end
	end
end

#prepare_input(length) ⇒ Input

Prepare the input stream which will be used for incoming data frames.

Returns:

  • (Input) —

    the input body.



97
98
99
100
101
102
103
 
# File 'lib/async/http/protocol/http2/stream.rb', line 97

def prepare_input(length)
	if @input.nil?
		@input = Input.new(self, length)
	else
		raise ArgumentError, "Input body already prepared!"
	end
end

#process_data(frame) ⇒ Object

Process an incoming DATA frame and write it to the input body.



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/async/http/protocol/http2/stream.rb', line 117

def process_data(frame)
	data = frame.unpack
	
	if input = @input
		unless data.empty?
			input.write(data)
		end
		
		if frame.end_stream?
			input.close_write
		end
	else
		# The application has closed the input, so discard incoming data while maintaining flow control for the stream.
		request_window_update
	end
	
	return data
rescue ::Protocol::HTTP2::ProtocolError
	raise
rescue # Anything else...
	send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR)
end

#process_headers(frame) ⇒ Object

Process an incoming HEADERS frame, dispatching to initial or trailing header handling.



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/async/http/protocol/http2/stream.rb', line 70

def process_headers(frame)
	if @headers and frame.end_stream?
		self.receive_trailing_headers(super, frame.end_stream?)
	else
		self.receive_initial_headers(super, frame.end_stream?)
	end
	
	if @input and frame.end_stream?
		@input.close_write
	end
rescue ::Protocol::HTTP::InvalidTrailerError => error
	Console.warn(self, error)
	
	send_reset_stream(::Protocol::HTTP2::Error::PROTOCOL_ERROR)
rescue ::Protocol::HTTP2::HeaderError => error
	Console.debug(self, "Error while processing headers!", error)
	
	send_reset_stream(error.code)
end

#receive_trailing_headers(headers, end_stream) ⇒ Object

Process trailing headers received after the body.



62
63
64
65
66
 
# File 'lib/async/http/protocol/http2/stream.rb', line 62

def receive_trailing_headers(headers, end_stream)
	headers.each do |key, value|
		add_header(key, value, trailer: true)
	end
end

#send_body(body, trailer = nil) ⇒ Object

Set the body and begin sending it.



157
158
159
160
161
 
# File 'lib/async/http/protocol/http2/stream.rb', line 157

def send_body(body, trailer = nil)
	@output = Output.new(self, body, trailer)
	
	@output.start
end

#send_data ⇒ Object

Send data and apply any pending application-side closure.



202
203
204
205
206
 
# File 'lib/async/http/protocol/http2/stream.rb', line 202

def send_data(...)
	result = super
	close_if_finished
	return result
end

#send_headers ⇒ Object

Send headers and apply any pending application-side closure.



195
196
197
198
199
 
# File 'lib/async/http/protocol/http2/stream.rb', line 195

def send_headers(...)
	result = super
	close_if_finished
	return result
end

#update_local_window(frame) ⇒ Object

Update the local flow control window after receiving data.



107
108
109
110
111
112
 
# File 'lib/async/http/protocol/http2/stream.rb', line 107

def update_local_window(frame)
	consume_local_window(frame)
	
	# This is done on demand in `Input#read`:
	# request_window_update
end

#wait_for_input ⇒ Object 



91
92
93
 
# File 'lib/async/http/protocol/http2/stream.rb', line 91

def wait_for_input
	return @input
end

#window_updated(size) ⇒ Object

Called when the flow control window is updated.



186
187
188
189
190
191
192
 
# File 'lib/async/http/protocol/http2/stream.rb', line 186

def window_updated(size)
	super
	
	@output&.window_updated(size)
	
	return true
end