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
- #add_trailer(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) ⇒ 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 |
#add_trailer(key, value) ⇒ Object
63 64 65 66 67 68 69 |
# File 'lib/async/http/protocol/http2/stream.rb', line 63 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 |
#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
161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 |
# File 'lib/async/http/protocol/http2/stream.rb', line 161 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.
141 142 143 144 145 146 147 148 149 |
# File 'lib/async/http/protocol/http2/stream.rb', line 141 def finish_output(error = nil) @output = nil if error send_reset_stream(::Protocol::HTTP2::Error::INTERNAL_ERROR) else send_data(nil, ::Protocol::HTTP2::END_STREAM) end end |
#prepare_input(length) ⇒ Input
Prepare the input stream which will be used for incoming data frames.
99 100 101 102 103 104 105 |
# File 'lib/async/http/protocol/http2/stream.rb', line 99 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
114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
# File 'lib/async/http/protocol/http2/stream.rb', line 114 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
77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 |
# File 'lib/async/http/protocol/http2/stream.rb', line 77 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 rescue ::Protocol::HTTP2::HeaderError => error Async.logger.error(self, error) send_reset_stream(error.code) end |
#receive_trailing_headers(headers, end_stream) ⇒ Object
71 72 73 74 75 |
# File 'lib/async/http/protocol/http2/stream.rb', line 71 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.
136 137 138 |
# File 'lib/async/http/protocol/http2/stream.rb', line 136 def send_body(body) @output = Output.for(self, body) end |
#update_local_window(frame) ⇒ Object
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
93 94 95 |
# File 'lib/async/http/protocol/http2/stream.rb', line 93 def wait_for_input return @input end |
#window_updated(size) ⇒ Object
151 152 153 154 155 |
# File 'lib/async/http/protocol/http2/stream.rb', line 151 def window_updated(size) super @output&.window_updated(size) end |