Methods
Public Class
Public Instance
Classes and Modules
Constants
| MAX_CONCURRENT_REQUESTS | = | ::HTTP2::DEFAULT_MAX_CONCURRENT_STREAMS |
Public Instance Aliases
| reset | -> | init_connection |
Public Class methods
new(buffer, options)
[show source]
# File lib/httpx/connection/http2.rb 53 def initialize(buffer, options) 54 @callbacks = @last_error = nil 55 @options = options 56 @settings = @options.http2_settings 57 @pending = [] 58 @streams = {} 59 @drains = {} 60 @pings = [] 61 @streams_to_close_after_receive = [] 62 @buffer = buffer 63 @handshake_completed = false 64 @wait_for_handshake = @settings.key?(:wait_for_handshake) ? @settings.delete(:wait_for_handshake) : true 65 @max_concurrent_requests = @options.max_concurrent_requests || MAX_CONCURRENT_REQUESTS 66 @max_requests = @options.max_requests 67 init_connection 68 end
Public Instance methods
<<(data)
[show source]
# File lib/httpx/connection/http2.rb 177 def <<(data) 178 @connection << data 179 180 while (stream, request, error = @streams_to_close_after_receive.shift) 181 # these streams were marked for cancellation due to errors found while processing the 182 # data received by the peer. 183 emit_stream_error(stream, request, error) 184 end 185 186 # if the connection transitioned to :closed, start tearing it down. Explicitly bypass the 187 # :closing transition to give it time to complete the inflight streams. 188 return unless @connection.state == :closed 189 190 close 191 end
close()
[show source]
# File lib/httpx/connection/http2.rb 135 def close 136 if @connection.closed? 137 # if there are inflight streams waiting to be completed AFTER the peer alread shut down the 138 # connection from their end, raise-or-retry depending on whether the error is recoverable. 139 if (ex = @last_error) 140 if @streams.any? || @pending.any? 141 if ex.is_a?(GoawayError) && !ex.unrecoverable? 142 inflight_unprocessed_requests = @streams.map do |req, _| 143 req.transition(:idle) 144 req 145 end 146 @streams.clear 147 @pending.unshift(*inflight_unprocessed_requests) 148 emit(:exhausted, ex) 149 else 150 while (req, _ = @streams.shift) 151 emit(:error, req, ex) 152 end 153 while (req = @pending.shift) 154 emit(:error, req, ex) 155 end 156 end 157 end 158 159 # clean pending payload now that everything's gone. 160 @buffer.clear 161 end 162 else 163 @connection.goaway 164 emit(:timeout, @options.timeout.close_handshake_timeout) 165 end 166 emit(:close) 167 end
consume()
[show source]
# File lib/httpx/connection/http2.rb 213 def consume 214 @streams.each do |request, stream| 215 next unless request.can_buffer? 216 217 handle(request, stream) 218 end 219 220 send_pending 221 end
empty?()
[show source]
# File lib/httpx/connection/http2.rb 169 def empty? 170 @connection.closed? || @streams.empty? 171 end
exhausted?()
[show source]
# File lib/httpx/connection/http2.rb 173 def exhausted? 174 !@max_requests.positive? 175 end
handle_error(ex, request = nil)
[show source]
# File lib/httpx/connection/http2.rb 223 def handle_error(ex, request = nil) 224 last_stream_id = Float::INFINITY 225 case ex 226 when OperationTimeoutError 227 if !@handshake_completed && @connection.state != :closed 228 # timed out waiting for HTTP/2 connection handshake to complete. 229 @connection.goaway(:settings_timeout, "closing due to settings timeout") 230 emit(:close_handshake) 231 settings_ex = SettingsTimeoutError.new(ex.timeout, ex.message) 232 settings_ex.set_backtrace(ex.backtrace) 233 ex = settings_ex 234 elsif @connection.state == :closing 235 # failed to receive data for inflight streams in time. 236 ex = @last_error || ex 237 end 238 when EOFError 239 if @connection.state == :closing 240 # socket closed while waiting for inflight streams to finish 241 ex = @last_error || ex 242 end 243 when GoawayError 244 last_stream_id = ex.last_stream_id 245 end 246 247 inflight_unprocessed_requests = [] #: Array[Request] 248 unprocessed_streams = [] #: Array[::HTTP2::Stream] 249 250 @streams.delete_if do |req, stream| 251 # allow inflight streams to finish 252 next(false) if !last_stream_id.infinite? && stream.id <= last_stream_id 253 254 @drains.delete(stream) 255 256 next(true) if request && request == req 257 258 if stream.id > last_stream_id 259 req.transition(:idle) 260 # unprocessed request 261 inflight_unprocessed_requests << req 262 unprocessed_streams << stream 263 264 next(true) 265 end 266 267 emit(:error, req, ex) 268 true 269 end 270 271 @pending.unshift(*inflight_unprocessed_requests) if inflight_unprocessed_requests.any? 272 273 unless @streams.empty? 274 unprocessed_streams.each(&:close) 275 276 # streams covered by the GOAWAY are still being processed. 277 return 278 end 279 280 if ex.is_a?(GoawayError) && !ex.unrecoverable? 281 # resend unprocessed requests on a different connection 282 emit(:exhausted, ex) if @pending.any? 283 return 284 end 285 286 while (req = @pending.shift) 287 next if request && request == req 288 289 emit(:error, req, ex) 290 end 291 end
interests()
[show source]
# File lib/httpx/connection/http2.rb 76 def interests 77 if @connection.state == :closed 78 return unless @handshake_completed 79 80 if @buffer.empty? 81 # a closed HTTP/2 connection can still receive DATA frames from inflight streams. 82 return :r unless @streams.empty? 83 84 return 85 end 86 87 # HTTP/2 GOAWAY frame buffered. 88 return :w 89 end 90 91 unless @connection.state == :connected && @handshake_completed 92 # HTTP/2 in intermediate state or still completing initialization- 93 return @buffer.empty? ? :r : :w 94 end 95 96 unless @connection.send_buffer.empty? 97 # HTTP/2 connection is buffering data chunks and failing to emit DATA frames, 98 # most likely because the flow control window is exhausted. 99 return :rw unless @buffer.empty? 100 101 # waiting for WINDOW_UPDATE frames 102 return :r 103 end 104 105 # there are pending bufferable requests 106 if !@pending.empty? && can_buffer_pending_requests? 107 # only wait for writable if not waiting on the peer. 108 return waiting_for_peer? ? :rw : :w 109 end 110 111 unless @drains.empty? 112 return :w unless @buffer.empty? 113 114 streams_with_no_window = @drains.none? do |stream, _| 115 # if an existing stream has buffered bytes, it must wait for 116 # WINDOW_UPDATE frames 117 stream.send_buffer.empty? 118 end 119 120 return streams_with_no_window ? :r : :rw 121 122 end 123 124 if @buffer.empty? 125 # skip if no more requests or pings to process 126 return unless waiting_for_peer? 127 128 :r 129 else 130 # buffered frames 131 waiting_for_peer? ? :rw : :w 132 end 133 end
ping()
[show source]
# File lib/httpx/connection/http2.rb 293 def ping 294 ping = SecureRandom.gen_random(8) 295 @connection.ping(ping.dup) 296 ensure 297 @pings << ping 298 end
send(request, head = false)
[show source]
# File lib/httpx/connection/http2.rb 193 def send(request, head = false) 194 unless can_buffer_pending_requests? 195 head ? @pending.unshift(request) : @pending << request 196 return false 197 end 198 unless (stream = @streams[request]) 199 stream = @connection.new_stream(**request.http2_stream_options) 200 handle_stream(stream, request) 201 @streams[request] = stream 202 @max_requests -= 1 203 end 204 handle(request, stream) 205 true 206 rescue ::HTTP2::Error::StreamLimitExceeded 207 @pending.unshift(request) 208 false 209 rescue ::HTTP2::Error::Error, ArgumentError => e 210 emit(:error, request, e) 211 end
timeout()
[show source]
# File lib/httpx/connection/http2.rb 70 def timeout 71 return @options.timeout.operation_timeout if @handshake_completed 72 73 @options.timeout.settings_timeout 74 end
waiting_for_ping?()
[show source]
# File lib/httpx/connection/http2.rb 300 def waiting_for_ping? 301 @pings.any? 302 end