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

close()

    # File lib/httpx/connection.rb
319 def close
320   transition(:active) if @state == :inactive
321 
322   @parser&.close
323 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
227 def connecting?
228   @state == :idle
229 end

deactivate()

    # File lib/httpx/connection.rb
437 def deactivate
438   transition(:inactive)
439 end

disconnect()

disconnects from the current session it’s attached to

    # File lib/httpx/connection.rb
472 def disconnect
473   return if @exhausted # it'll reset
474 
475   return unless (current_session = @current_session) && (current_selector = @current_selector)
476 
477   @current_session = @current_selector = nil
478 
479   current_session.deselect_connection(self, current_selector, @cloned)
480 end

force_close(delete_pending = false)

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

    # File lib/httpx/connection.rb
344 def force_close(delete_pending = false)
345   force_purge
346   return unless @state == :closed
347 
348   if delete_pending
349     @pending.clear
350   elsif (parser = @parser)
351     enqueue_pending_requests_from_parser(parser)
352     return unless @pending.empty?
353   end
354 
355   disconnect
356   emit(:force_closed, delete_pending)
357 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
361 def force_reset(cloned = false)
362   @state = :closing
363   @cloned = cloned
364   transition(:closed)
365 end

handle_connect_error(error)

    # File lib/httpx/connection.rb
463 def handle_connect_error(error)
464   return on_error(error) unless @sibling && @sibling.connecting?
465 
466   @sibling.merge(self)
467 
468   force_reset(true)
469 end

handle_socket_timeout(interval)

    # File lib/httpx/connection.rb
445 def handle_socket_timeout(interval)
446   error = OperationTimeoutError.new(interval, "timed out while waiting on select")
447   error.set_backtrace(caller)
448   on_error(error)
449 end

idling()

    # File lib/httpx/connection.rb
420 def idling
421   purge_after_closed
422 
423   return unless @state == :closed
424 
425   @write_buffer.clear
426   transition(:idle)
427   return unless @parser
428 
429   enqueue_pending_requests_from_parser(@parser)
430   @parser = nil
431 end

inflight?()

    # File lib/httpx/connection.rb
231 def inflight?
232   @parser && (
233     # parser may be dealing with other requests (possibly started from a different fiber)
234     !@parser.empty? ||
235     # connection may be doing connection termination handshake
236     !@write_buffer.empty?
237   )
238 end

initial_call()

    # File lib/httpx/connection.rb
315 def initial_call
316   call
317 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
525 def inspect
526   "#<#{self.class}:#{object_id} " \
527     "@origin=#{@origin} " \
528     "@state=#{@state} " \
529     "@pending=#{@pending.size} " \
530     "@io=#{@io}>"
531 end

interests()

    # File lib/httpx/connection.rb
240 def interests
241   # connecting
242   if connecting?
243     connect
244 
245     return @io.interests if connecting?
246   end
247 
248   @parser&.interests
249 rescue Error => e
250   on_error(e)
251   nil
252 end

io_connected?()

    # File lib/httpx/connection.rb
221 def io_connected?
222   return @coalesced_connection.io_connected? if @coalesced_connection
223 
224   @io && @io.state == :connected
225 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
482 def on_connect_error(e)
483   # connect errors, exit gracefully
484   error = ConnectionError.new(e.message)
485   error.set_backtrace(e.backtrace)
486   handle_connect_error(error) if connecting?
487   force_close
488 end

on_error(error, request = nil)

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

on_io_error(e)

    # File lib/httpx/connection.rb
490 def on_io_error(e)
491   on_error(e)
492 
493   # do not force close if parser resets the connection.
494   # can happen i.e. when HTTP/1.1 pipelining is disabled.
495   return if @state == :idle && @pending.any?
496 
497   force_close(true)
498 end

open?()

    # File lib/httpx/connection.rb
441 def open?
442   @state == :open || @state == :inactive
443 end

peer()

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

pending_count()

returns the number of pending requests both in the connection as well as parser pending queues

    # File lib/httpx/connection.rb
200 def pending_count
201   count = @pending.size
202   count += @parser.pending.size if @parser.respond_to?(:pending)
203   count
204 end

purge_pending(&block)

    # File lib/httpx/connection.rb
206 def purge_pending(&block)
207   if @parser
208     pending = @parser.pending
209     @inflight -= pending.size
210     pending.reject! do |req|
211       block.call(req)
212       true
213     end
214   end
215   @pending.reject! do |req|
216     block.call(req)
217     true
218   end
219 end

reset()

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

send(request)

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

sibling=(connection)

    # File lib/httpx/connection.rb
451 def sibling=(connection)
452   @sibling = connection
453 
454   return unless connection
455 
456   @main_sibling = connection.sibling.nil?
457 
458   return unless @main_sibling
459 
460   connection.sibling = self
461 end

terminate()

    # File lib/httpx/connection.rb
325 def terminate
326   case @state
327   when :idle
328     purge_after_closed
329 
330     # @fiber-switch-guard
331     if @io.can_disconnect? && @pending.empty?
332       disconnect
333       return
334     end
335   when :closed
336     @connected_at = nil
337   end
338 
339   close
340 end

timeout()

    # File lib/httpx/connection.rb
412 def timeout
413   return if @state == :closed || @state == :inactive
414 
415   return @timeout if @timeout
416 
417   @options.timeout[:operation_timeout]
418 end

to_io()

    # File lib/httpx/connection.rb
254 def to_io
255   @io.to_io
256 end

used?()

    # File lib/httpx/connection.rb
433 def used?
434   @connected_at
435 end