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

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

Direct Known Subclasses

Request::Stream, Response::Stream

Defined Under Namespace

Classes: Input, Output

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Stream

Returns a new instance of Stream.



158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/async/http/protocol/http2/stream.rb', line 158

def initialize(*)
  super
  
  @headers = nil
  @trailers = nil
  
  # Input buffer, reading request body, or response body (receive_data):
  @length = nil
  @input = nil
  
  # Output buffer, writing request body or response body (window_updated):
  @output = nil
end

Instance Attribute Details

#headers ⇒ Object

Returns the value of attribute headers.



172
173
174
 
# File 'lib/async/http/protocol/http2/stream.rb', line 172

def headers
  @headers
end

#input ⇒ Object (readonly)

Returns the value of attribute input.



174
175
176
 
# File 'lib/async/http/protocol/http2/stream.rb', line 174

def input
  @input
end

Instance Method Details

#add_header(key, value) ⇒ Object 



176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/async/http/protocol/http2/stream.rb', line 176

def add_header(key, value)
  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)
  end
end

#add_trailer(key, value) ⇒ Object 



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

def add_trailer(key, value)
  if @trailers.include(key)
    add_header(key, value)
  else
    raise ::Protocol::HTTP2::HeaderError, "Cannot add trailer #{key} as it was not specified in trailers!"
  end
end

#close(error = nil) ⇒ Object 



271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/async/http/protocol/http2/stream.rb', line 271

def close(error = nil)
  super
  
  if @input
    @input.close(error)
    @input = nil
  end
  
  if @output
    @output.stop(error)
    @output = nil
  end
end

#prepare_input(length) ⇒ Input

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

Returns:

  • (Input) —

    the input body.



224
225
226
227
228
229
230
 
# File 'lib/async/http/protocol/http2/stream.rb', line 224

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 



239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/async/http/protocol/http2/stream.rb', line 239

def process_data(frame)
  data = frame.unpack
  
  if @input
    unless data.empty?
      @input.write(data)
    end
    
    if frame.end_stream?
      @input.close
      @input = nil
    end
  end
  
  return data
rescue ::Protocol::HTTP2::ProtocolError
  raise
rescue # Anything else...
  send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR)
end

#receive_headers(frame) ⇒ Object 



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/async/http/protocol/http2/stream.rb', line 202

def receive_headers(frame)
  if @headers.nil?
    @headers = ::Protocol::HTTP::Headers.new
    self.receive_initial_headers(super, frame.end_stream?)
    @trailers = @headers[TRAILERS]
  elsif @trailers and frame.end_stream?
    self.receive_trailing_headers(super, frame.end_stream?)
  else
    raise ::Protocol::HTTP2::HeaderError, "Unable to process headers!"
  end
rescue ::Protocol::HTTP2::HeaderError => error
  Async.logger.error(self, error)
  
  send_reset_stream(error.code)
end

#receive_trailing_headers(headers, end_stream) ⇒ Object 



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

def receive_trailing_headers(headers, end_stream)
  headers.each do |key, value|
    add_trailer(key, value)
  end
end

#send_body(body) ⇒ Object

Set the body and begin sending it.



261
262
263
 
# File 'lib/async/http/protocol/http2/stream.rb', line 261

def send_body(body)
  @output = Output.for(self, body)
end

#update_local_window(frame) ⇒ Object 



232
233
234
235
236
237
 
# File 'lib/async/http/protocol/http2/stream.rb', line 232

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 



218
219
220
 
# File 'lib/async/http/protocol/http2/stream.rb', line 218

def wait_for_input
  return @input
end

#window_updated(size) ⇒ Object 



265
266
267
268
269
 
# File 'lib/async/http/protocol/http2/stream.rb', line 265

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