Class: Async::HTTP::Protocol::HTTP2::Stream
- Inherits:
-
Protocol::HTTP2::Stream
- Object
- Protocol::HTTP2::Stream
- Async::HTTP::Protocol::HTTP2::Stream
- Defined in:
- lib/async/http/protocol/http2/stream.rb
Direct Known Subclasses
Instance Attribute Summary collapse
-
#headers ⇒ Object
Returns the value of attribute headers.
-
#input ⇒ Object
readonly
Returns the value of attribute input.
Instance Method Summary collapse
- #add_header(key, value) ⇒ Object
-
#closed(error) ⇒ Object
When the stream transitions to the closed state, this method is called.
-
#finish_output(error = nil) ⇒ Object
Called when the output terminates normally.
-
#initialize ⇒ Stream
constructor
A new instance of Stream.
-
#prepare_input(length) ⇒ Input
Prepare the input stream which will be used for incoming data frames.
- #process_data(frame) ⇒ Object
- #process_headers(frame) ⇒ Object
- #receive_trailing_headers(headers, end_stream) ⇒ Object
-
#send_body(body, trailers = nil) ⇒ Object
Set the body and begin sending it.
- #update_local_window(frame) ⇒ Object
- #wait_for_input ⇒ Object
- #window_updated(size) ⇒ Object
Constructor Details
#initialize ⇒ Stream
Returns a new instance of Stream.
33 34 35 36 37 38 39 40 41 42 43 44 45 |
# File 'lib/async/http/protocol/http2/stream.rb', line 33 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.
47 48 49 |
# File 'lib/async/http/protocol/http2/stream.rb', line 47 def headers @headers end |
#input ⇒ Object (readonly)
Returns the value of attribute input.
49 50 51 |
# File 'lib/async/http/protocol/http2/stream.rb', line 49 def input @input end |
Instance Method Details
#add_header(key, value) ⇒ Object
51 52 53 54 55 56 57 58 59 60 61 |
# File 'lib/async/http/protocol/http2/stream.rb', line 51 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 |
#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
173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 |
# File 'lib/async/http/protocol/http2/stream.rb', line 173 def closed(error) super if @input @input.close(error) @input = nil end if @output @output.stop(error) @output = nil end return self end |
#finish_output(error = nil) ⇒ Object
Called when the output terminates normally.
146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
# File 'lib/async/http/protocol/http2/stream.rb', line 146 def finish_output(error = nil) trailers = @output&.trailers @output = nil if error send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) else # Write trailers? if trailers send_headers(nil, trailers, ::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.
102 103 104 105 106 107 108 |
# File 'lib/async/http/protocol/http2/stream.rb', line 102 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
117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
# File 'lib/async/http/protocol/http2/stream.rb', line 117 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 |
#process_headers(frame) ⇒ Object
73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
# File 'lib/async/http/protocol/http2/stream.rb', line 73 def process_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 # TODO this might need to be in an ensure block: if @input and frame.end_stream? @input.close($!) @input = nil end rescue ::Protocol::HTTP2::HeaderError => error Async.logger.error(self, error) send_reset_stream(error.code) end |
#receive_trailing_headers(headers, end_stream) ⇒ Object
63 64 65 66 67 68 69 70 71 |
# File 'lib/async/http/protocol/http2/stream.rb', line 63 def receive_trailing_headers(headers, end_stream) headers.each do |key, value| if @trailers.include?(key) add_header(key, value) else raise ::Protocol::HTTP2::HeaderError, "Cannot add trailer #{key} as it was not specified as a trailer!" end end end |
#send_body(body, trailers = nil) ⇒ Object
Set the body and begin sending it.
139 140 141 142 143 |
# File 'lib/async/http/protocol/http2/stream.rb', line 139 def send_body(body, trailers = nil) @output = Output.new(self, body, trailers) @output.start end |
#update_local_window(frame) ⇒ Object
110 111 112 113 114 115 |
# File 'lib/async/http/protocol/http2/stream.rb', line 110 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
96 97 98 |
# File 'lib/async/http/protocol/http2/stream.rb', line 96 def wait_for_input return @input end |
#window_updated(size) ⇒ Object
163 164 165 166 167 |
# File 'lib/async/http/protocol/http2/stream.rb', line 163 def window_updated(size) super @output&.window_updated(size) end |