HTTPX::Connection

The Connection can be watched for IO events.

It contains the io object to read/write from, and knows what to do when it can.

It defers connecting until absolutely necessary. Connection should be triggered from the IO selector (until then, any request will be queued).

A connection boots up its parser after connection is established. All pending requests will be redirected there after connection.

A connection can be prevented from closing by the parser, that is, if there are pending requests. This will signal that the connection was prematurely closed, due to a possible number of conditions:

Remote peer closed the connection (“Connection: close”);
Remote peer doesn’t support pipelining;

A connection may also route requests for a different host for which the io was connected to, provided that the IP is the same and the port and scheme as well. This will allow to share the same socket to send HTTP/2 requests to different hosts.

Methods

Public Class

new

Public Instance

Protected Instance

read_buffer
sibling
ssl_session
write_buffer

Included modules

Classes and Modules

Attributes

 current_selector
 current_session
 family
 io
 options
 origin
 origins
 pending
 read_buffer
 sibling
 ssl_session
 state
 type
 write_buffer

Public Class methods

new(uri, options)

   # File lib/httpx/connection.rb
48 def initialize(uri, options)
49   @current_session = @current_selector = @max_concurrent_requests =
50                        @parser = @sibling = @coalesced_connection = @altsvc_connection =
51                                               @callbacks = @ping_timer = @family =
52                                                              @io = @ssl_session = @timeout = @connected_at = @response_received_at = nil
53 
54   @exhausted = @cloned = @main_sibling =
55                  @reset_timeout_on_call =
56                    # variable used to gate against a potential endless loop where the peer continuously closes the connection with
57                    # GOAWAY frames without ever processing a request.
58                    @previously_exhausted_with_error = false
59 
60   @options = Options.new(options)
61   @type = initialize_type(uri, @options)
62   @origins = [uri.origin]
63   @origin = Utils.to_uri(uri.origin)
64   @window_size = @options.window_size
65   @read_buffer = Buffer.new(@options.buffer_size)
66   @write_buffer = Buffer.new(@options.buffer_size)
67   @pending = []
68   @inflight = 0
69   @keep_alive_timeout = @options.timeout[:keep_alive_timeout]
70   @no_more_requests_counter = 0
71 
72   if @options.io
73     # if there's an already open IO, get its
74     # peer address, and force-initiate the parser
75     transition(:already_open)
76     @io = build_socket
77     parser
78   else
79     transition(:idle)
80   end
81 
82   case @options.addresses
83   when Array
84     self.addresses = @options.addresses
85   when Hash
86     # TODO: && @origin.host not an IP
87     self.addresses = @options.addresses[@origin.host]
88   end
89 end

Public Instance methods

addresses()

    # File lib/httpx/connection.rb
128 def addresses
129   @io && @io.addresses
130 end

addresses=(addrs)

this is a semi-private method, to be used by the resolver to initiate the io object.

    # File lib/httpx/connection.rb
120 def addresses=(addrs)
121   if @io
122     @io.add_addresses(addrs)
123   else
124     @io = build_socket(addrs)
125   end
126 end

addresses?()

    # File lib/httpx/connection.rb
132 def addresses?
133   @io && @io.addresses?
134 end

call()

    # File lib/httpx/connection.rb
250 def call
251   if @reset_timeout_on_call
252     @timeout = nil
253     @reset_timeout_on_call = false
254   end
255 
256   case @state
257   when :idle
258     return if no_more_requests?
259 
260     connect
261 
262     # when opening the tcp or ssl socket fails
263     return if @state == :closed
264 
265     consume
266   when :closed
267     return if no_more_requests?
268 
269     # there are pending requests to send, restart the state machine.
270     idling
271 
272     # @fiber-switch-guard
273     # fiber may have switch after ensuring that @io is closed.
274     return unless @state == :idle
275 
276     call
277   when :closing
278     consume
279     transition(:closed)
280 
281     # @fiber-switch-guard
282     # fiber may have switch while closing @io.
283     return if @state == :closed &&
284               # only remain here if there are pending requests.
285               @pending.empty?
286 
287     call
288   when :open
289     consume
290   end
291   nil
292 rescue Errno::ECONNRESET,
293        Errno::EINVAL,
294        SocketError,
295        IOError,
296        TLSError => e
297   @write_buffer.clear
298   on_io_error(e)
299 rescue Error => e
300   @write_buffer.clear
301   on_error(e)
302 rescue Exception => e # rubocop:disable Lint/RescueException
303   force_close(true)
304   raise e
305 end

close()

    # File lib/httpx/connection.rb
311 def close
312   transition(:active) if @state == :inactive
313 
314   @parser&.close
315 end

coalescable?(connection)

coalescable connections need to be mergeable! but internally, mergeable? is called before coalescable?

    # File lib/httpx/connection.rb
173 def coalescable?(connection)
174   if @io.protocol == "h2" &&
175      @origin.scheme == "https" &&
176      connection.origin.scheme == "https" &&
177      @io.can_verify_peer?
178     @io.verify_hostname(connection.origin.host)
179   else
180     @origin == connection.origin
181   end
182 end

coalesce!(connection)

coalesces self into connection.

    # File lib/httpx/connection.rb
160 def coalesce!(connection)
161   @coalesced_connection = connection
162 
163   close_sibling
164   connection.merge(self)
165 end

coalesced?()

    # File lib/httpx/connection.rb
167 def coalesced?
168   @coalesced_connection
169 end

connecting?()

    # File lib/httpx/connection.rb
219 def connecting?
220   @state == :idle
221 end

deactivate()

    # File lib/httpx/connection.rb
429 def deactivate
430   transition(:inactive)
431 end

disconnect()

disconnects from the current session it’s attached to

    # File lib/httpx/connection.rb
464 def disconnect
465   return if @exhausted # it'll reset
466 
467   return unless (current_session = @current_session) && (current_selector = @current_selector)
468 
469   @current_session = @current_selector = nil
470 
471   current_session.deselect_connection(self, current_selector, @cloned)
472 end

force_close(delete_pending = false)

bypasses state machine rules while setting the connection in the :closed state.

    # File lib/httpx/connection.rb
336 def force_close(delete_pending = false)
337   force_purge
338   return unless @state == :closed
339 
340   if delete_pending
341     @pending.clear
342   elsif (parser = @parser)
343     enqueue_pending_requests_from_parser(parser)
344     return unless @pending.empty?
345   end
346 
347   disconnect
348   emit(:force_closed, delete_pending)
349 end

force_reset(cloned = false)

bypasses the state machine to force closing of connections still connecting. only used for Happy Eyeballs v2.

    # File lib/httpx/connection.rb
353 def force_reset(cloned = false)
354   @state = :closing
355   @cloned = cloned
356   transition(:closed)
357 end

handle_connect_error(error)

    # File lib/httpx/connection.rb
455 def handle_connect_error(error)
456   return on_error(error) unless @sibling && @sibling.connecting?
457 
458   @sibling.merge(self)
459 
460   force_reset(true)
461 end

handle_socket_timeout(interval)

    # File lib/httpx/connection.rb
437 def handle_socket_timeout(interval)
438   error = OperationTimeoutError.new(interval, "timed out while waiting on select")
439   error.set_backtrace(caller)
440   on_error(error)
441 end

idling()

    # File lib/httpx/connection.rb
412 def idling
413   purge_after_closed
414 
415   return unless @state == :closed
416 
417   @write_buffer.clear
418   transition(:idle)
419   return unless @parser
420 
421   enqueue_pending_requests_from_parser(@parser)
422   @parser = nil
423 end

inflight?()

    # File lib/httpx/connection.rb
223 def inflight?
224   @parser && (
225     # parser may be dealing with other requests (possibly started from a different fiber)
226     !@parser.empty? ||
227     # connection may be doing connection termination handshake
228     !@write_buffer.empty?
229   )
230 end

initial_call()

    # File lib/httpx/connection.rb
307 def initial_call
308   call
309 end

initialize_dup(orig)

dupped initialization

    # File lib/httpx/connection.rb
 92 def initialize_dup(orig)
 93   super
 94   @callbacks = @parser = @sibling = @coalesced_connection = @altsvc_connection = nil
 95   @origins = orig.origins.dup
 96   @read_buffer = orig.read_buffer.dup
 97   @write_buffer = orig.write_buffer.dup
 98   @inflight = 0
 99   @pending = []
100   transition(:idle)
101 
102   if @io
103     # initialize new IO object with the same set of addresses
104     addresses = @io.addresses
105     @io = nil
106     self.addresses = addresses
107   end
108 
109   return unless @current_session && @current_selector
110 
111   @current_session.pin(self, @current_selector)
112 end

inspect()

simplecov:disable

    # File lib/httpx/connection.rb
517 def inspect
518   "#<#{self.class}:#{object_id} " \
519     "@origin=#{@origin} " \
520     "@state=#{@state} " \
521     "@pending=#{@pending.size} " \
522     "@io=#{@io}>"
523 end

interests()

    # File lib/httpx/connection.rb
232 def interests
233   # connecting
234   if connecting?
235     connect
236 
237     return @io.interests if connecting?
238   end
239 
240   @parser&.interests
241 rescue Error => e
242   on_error(e)
243   nil
244 end

io_connected?()

    # File lib/httpx/connection.rb
213 def io_connected?
214   return @coalesced_connection.io_connected? if @coalesced_connection
215 
216   @io && @io.state == :connected
217 end

match?(uri, options)

    # File lib/httpx/connection.rb
136 def match?(uri, options)
137   return false if !used? && (@state == :closing || @state == :closed)
138 
139   @origins.include?(uri.origin) &&
140     # if there is more than one origin to match, it means that this connection
141     # was the result of coalescing. To prevent blind trust in the case where the
142     # origin came from an ORIGIN frame, we're going to verify the hostname with the
143     # SSL certificate
144     (@origins.size == 1 || @origin == uri.origin || (@io.is_a?(SSL) && @io.verify_hostname(uri.host))) &&
145     @options.connection_options_match?(options)
146 end

merge(connection)

    # File lib/httpx/connection.rb
184 def merge(connection)
185   @origins |= connection.instance_variable_get(:@origins)
186   if @ssl_session.nil? && (ssl_session = connection.ssl_session)
187     @ssl_session = ssl_session
188     # the socket only needs the merged session if it can still resume it,
189     # i.e. if TLS hasn't been negotiated yet.
190     @io.ssl_session = ssl_session if @io.is_a?(SSL) && !@io.connected?
191   end
192   connection.purge_pending do |req|
193     req.transition(:idle)
194     send(req)
195   end
196 end

mergeable?(connection)

    # File lib/httpx/connection.rb
148 def mergeable?(connection)
149   return false if @state == :closing || @state == :closed || !@io
150 
151   return false unless connection.addresses
152 
153   (
154     (open? && @origin == connection.origin) ||
155     !(@io.addresses & (connection.addresses || [])).empty?
156   ) && @options.connection_options_match?(connection.options)
157 end

on_connect_error(e)

    # File lib/httpx/connection.rb
474 def on_connect_error(e)
475   # connect errors, exit gracefully
476   error = ConnectionError.new(e.message)
477   error.set_backtrace(e.backtrace)
478   handle_connect_error(error) if connecting?
479   force_close
480 end

on_error(error, request = nil)

    # File lib/httpx/connection.rb
492 def on_error(error, request = nil)
493   if error.is_a?(OperationTimeoutError)
494     # inactive connections do not contribute to the select loop, therefore
495     # they should not fail due to such errors.
496     return if @state == :inactive
497 
498     if (current_timeout = @timeout || timeout)
499       # if @timeout isn't set, it's not a completion-based timeout, i.e.
500       # operation timeout; in such a case, we have to reset it once it piggybacks.
501       @reset_timeout_on_call ||= @timeout.nil?
502 
503       @timeout = current_timeout - error.timeout
504       return unless @timeout <= 0
505 
506       @timeout = nil
507       @reset_timeout_on_call = false
508     end
509 
510     error = error.to_connection_error if connecting?
511   end
512   handle_error(error, request)
513   reset
514 end

on_io_error(e)

    # File lib/httpx/connection.rb
482 def on_io_error(e)
483   on_error(e)
484 
485   # do not force close if parser resets the connection.
486   # can happen i.e. when HTTP/1.1 pipelining is disabled.
487   return if @state == :idle && @pending.any?
488 
489   force_close(true)
490 end

open?()

    # File lib/httpx/connection.rb
433 def open?
434   @state == :open || @state == :inactive
435 end

peer()

    # File lib/httpx/connection.rb
114 def peer
115   @origin
116 end

purge_pending(&block)

    # File lib/httpx/connection.rb
198 def purge_pending(&block)
199   if @parser
200     pending = @parser.pending
201     @inflight -= pending.size
202     pending.reject! do |req|
203       block.call(req)
204       true
205     end
206   end
207   @pending.reject! do |req|
208     block.call(req)
209     true
210   end
211 end

reset()

    # File lib/httpx/connection.rb
359 def reset
360   return if @state == :closing || @state == :closed
361 
362   # do not reset a connection which may have restarted back to :idle, such when the parser resets
363   # (example: HTTP/1 parser disabling pipelining)
364   return if @state == :idle && @pending.any?
365 
366   reset_ping_timer
367 
368   parser = @parser
369 
370   if parser && parser.respond_to?(:max_concurrent_requests)
371     # if connection being reset has at some downgraded the number of concurrent
372     # requests, such as in the case where an attempt to use HTTP/1 pipelining failed,
373     # keep that information around.
374     @max_concurrent_requests = parser.max_concurrent_requests
375   end
376 
377   transition(:closing)
378   transition(:closed)
379 end

send(request)

    # File lib/httpx/connection.rb
381 def send(request)
382   return @coalesced_connection.send(request) if @coalesced_connection
383 
384   if @parser && !@write_buffer.full?
385     if @response_received_at && @keep_alive_timeout &&
386        Utils.elapsed_time(@response_received_at) > @keep_alive_timeout
387       # when pushing a request into an existing connection, we have to check whether there
388       # is the possibility that the connection might have extended the keep alive timeout.
389       # for such cases, we want to ping for availability before deciding to shovel requests.
390       log(level: 3) { "keep alive timeout expired, pinging connection..." }
391       @pending << request
392       transition(:active) if @state == :inactive
393       request.ping!
394       ping(request)
395       return
396     end
397 
398     send_request_to_parser(request)
399   else
400     @pending << request
401   end
402 end

sibling=(connection)

    # File lib/httpx/connection.rb
443 def sibling=(connection)
444   @sibling = connection
445 
446   return unless connection
447 
448   @main_sibling = connection.sibling.nil?
449 
450   return unless @main_sibling
451 
452   connection.sibling = self
453 end

terminate()

    # File lib/httpx/connection.rb
317 def terminate
318   case @state
319   when :idle
320     purge_after_closed
321 
322     # @fiber-switch-guard
323     if @io.can_disconnect? && @pending.empty?
324       disconnect
325       return
326     end
327   when :closed
328     @connected_at = nil
329   end
330 
331   close
332 end

timeout()

    # File lib/httpx/connection.rb
404 def timeout
405   return if @state == :closed || @state == :inactive
406 
407   return @timeout if @timeout
408 
409   @options.timeout[:operation_timeout]
410 end

to_io()

    # File lib/httpx/connection.rb
246 def to_io
247   @io.to_io
248 end

used?()

    # File lib/httpx/connection.rb
425 def used?
426   @connected_at
427 end