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

Public Instance methods

addresses()

    # File lib/httpx/connection.rb
126 def addresses
127   @io && @io.addresses
128 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
118 def addresses=(addrs)
119   if @io
120     @io.add_addresses(addrs)
121   else
122     @io = build_socket(addrs)
123   end
124 end

addresses?()

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

call()

    # File lib/httpx/connection.rb
265 def call
266   if @reset_timeout_on_call
267     @timeout = nil
268     @reset_timeout_on_call = false
269   end
270 
271   case @state
272   when :idle
273     return if no_more_requests?
274 
275     try_connect
276 
277     # when opening the tcp or ssl socket fails
278     return if @state == :closed
279 
280     consume
281   when :closed
282     return if no_more_requests?
283 
284     # there are pending requests to send, restart the state machine.
285     idling
286 
287     # @fiber-switch-guard
288     # fiber may have switch after ensuring that @io is closed.
289     return unless @state == :idle
290 
291     call
292   when :closing
293     # if bytes are available, that's the client GOAWAY frame. Otherwise, the peer closed it.
294     sending_goaway = !@write_buffer.empty?
295 
296     consume
297 
298     if sending_goaway
299       # parser may have inflight streams which need terminating.
300       @parser&.close
301       transition(:closed)
302     elsif @state == :closing
303       return
304     end
305 
306     # @fiber-switch-guard
307     # fiber may have switch while closing @io.
308     return if @state == :closed &&
309               # only remain here if there are pending requests.
310               @pending.empty?
311 
312     call
313   when :open
314     consume
315   end
316   nil
317 rescue Errno::ECONNRESET,
318        Errno::EINVAL,
319        SocketError,
320        IOError,
321        TLSError => e
322   @write_buffer.clear
323   on_io_error(e)
324 rescue Error => e
325   @write_buffer.clear
326   on_error(e)
327 rescue Exception => e # rubocop:disable Lint/RescueException
328   force_close(true)
329   raise e
330 end

close()

    # File lib/httpx/connection.rb
336 def close
337   transition(:active) if @state == :inactive
338 
339   @parser&.close
340 end

closed_for_select?()

whether the connection can be removed from the selector.

    # File lib/httpx/connection.rb
234 def closed_for_select?
235   @state == :closed
236 end

coalescable?(connection)

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

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

coalesce!(connection)

coalesces self into connection.

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

coalesced?()

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

connecting?()

    # File lib/httpx/connection.rb
229 def connecting?
230   @state == :idle
231 end

deactivate()

    # File lib/httpx/connection.rb
464 def deactivate
465   transition(:inactive)
466 end

disconnect()

disconnects from the current session it’s attached to

    # File lib/httpx/connection.rb
499 def disconnect
500   return if @exhausted # it'll reset
501 
502   return unless (current_session = @current_session) && (current_selector = @current_selector)
503 
504   @current_session = @current_selector = nil
505 
506   current_session.deselect_connection(self, current_selector, @cloned)
507 end

force_close(delete_pending = false)

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

    # File lib/httpx/connection.rb
362 def force_close(delete_pending = false)
363   force_purge
364   return unless @state == :closed
365 
366   if delete_pending
367     @pending.clear
368   elsif (parser = @parser)
369     enqueue_pending_requests_from_parser(parser)
370     return unless @pending.empty?
371   end
372 
373   disconnect
374   emit(:force_closed, delete_pending)
375 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
379 def force_reset(cloned = false)
380   @state = :closing
381   @cloned = cloned
382   transition(:closed)
383 end

handle_connect_error(error)

    # File lib/httpx/connection.rb
490 def handle_connect_error(error)
491   return on_error(error) unless @sibling && @sibling.connecting?
492 
493   @sibling.merge(self)
494 
495   force_reset(true)
496 end

handle_socket_timeout(interval)

    # File lib/httpx/connection.rb
472 def handle_socket_timeout(interval)
473   error = OperationTimeoutError.new(interval, "timed out while waiting on select")
474   error.set_backtrace(caller)
475   on_error(error)
476 end

idling()

    # File lib/httpx/connection.rb
447 def idling
448   purge_after_closed
449 
450   return unless @state == :closed
451 
452   @write_buffer.clear
453   transition(:idle)
454   return unless @parser
455 
456   enqueue_pending_requests_from_parser(@parser)
457   @parser = nil
458 end

inflight?()

    # File lib/httpx/connection.rb
238 def inflight?
239   @parser && (
240     # parser may be dealing with other requests (possibly started from a different fiber)
241     !@parser.empty? ||
242     # connection may be doing connection termination handshake
243     !@write_buffer.empty?
244   )
245 end

initial_call()

    # File lib/httpx/connection.rb
332 def initial_call
333   call
334 end

initialize_dup(orig)

dupped initialization

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

inspect()

simplecov:disable

    # File lib/httpx/connection.rb
566 def inspect
567   "#<#{self.class}:#{object_id} " \
568     "@origin=#{@origin} " \
569     "@state=#{@state} " \
570     "@pending=#{@pending.size} " \
571     "@io=#{@io}>"
572 end

interests()

    # File lib/httpx/connection.rb
247 def interests
248   # connecting
249   if connecting?
250     try_connect
251 
252     return @io.interests if connecting?
253   end
254 
255   @parser&.interests
256 rescue Error => e
257   on_error(e)
258   nil
259 end

io_connected?()

    # File lib/httpx/connection.rb
223 def io_connected?
224   return @coalesced_connection.io_connected? if @coalesced_connection
225 
226   @io && @io.state == :connected
227 end

match?(uri, options)

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

merge(connection)

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

mergeable?(connection)

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

on_connect_error(e)

    # File lib/httpx/connection.rb
509 def on_connect_error(e)
510   # connect errors, exit gracefully
511   error = ConnectionError.new(e.message)
512   error.set_backtrace(e.backtrace)
513   handle_connect_error(error) if connecting?
514   force_close
515 end

on_error(error, request = nil)

    # File lib/httpx/connection.rb
527 def on_error(error, request = nil)
528   if error.is_a?(OperationTimeoutError)
529     # inactive connections do not contribute to the select loop, therefore
530     # they should not fail due to such errors.
531     return if @state == :inactive
532 
533     attempt_timeout = connect_attempt_time_left if connecting?
534 
535     current_timeout = @timeout
536     current_timeout ||= @options.timeout.operation_timeout unless @state == :closed
537 
538     if current_timeout
539       # if @timeout isn't set, it's not a completion-based timeout, i.e.
540       # operation timeout; in such a case, we have to reset it once it piggybacks.
541       @reset_timeout_on_call ||= @timeout.nil?
542 
543       @timeout = current_timeout - error.timeout
544 
545       if @timeout.positive?
546         # if this was a connect timeout error and there are more addresses to try.
547         connect_next(error) if attempt_timeout && !attempt_timeout.positive?
548         return
549       end
550 
551       @timeout = nil
552       @reset_timeout_on_call = false
553     elsif attempt_timeout
554       # when request timeouts are set instead.
555       connect_next(error) unless attempt_timeout.positive?
556       return
557     end
558 
559     error = error.to_connection_error if connecting?
560   end
561   handle_error(error, request)
562   reset
563 end

on_io_error(e)

    # File lib/httpx/connection.rb
517 def on_io_error(e)
518   on_error(e)
519 
520   # do not force close if parser resets the connection.
521   # can happen i.e. when HTTP/1.1 pipelining is disabled.
522   return if @state == :idle && @pending.any?
523 
524   force_close(true)
525 end

open?()

    # File lib/httpx/connection.rb
468 def open?
469   @state == :open || @state == :inactive
470 end

peer()

    # File lib/httpx/connection.rb
112 def peer
113   @origin
114 end

pending_count()

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

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

purge_pending(&block)

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

reset()

    # File lib/httpx/connection.rb
385 def reset
386   return if @state == :closed
387 
388   # do not reset a connection which may have restarted back to :idle, such when the parser resets
389   # (example: HTTP/1 parser disabling pipelining)
390   return if @state == :idle && @pending.any?
391 
392   reset_ping_timer
393 
394   parser = @parser
395 
396   if parser && parser.respond_to?(:max_concurrent_requests)
397     # if connection being reset has at some downgraded the number of concurrent
398     # requests, such as in the case where an attempt to use HTTP/1 pipelining failed,
399     # keep that information around.
400     @max_concurrent_requests = parser.max_concurrent_requests
401   end
402 
403   transition(:closing)
404   transition(:closed)
405 end

send(request)

    # File lib/httpx/connection.rb
407 def send(request)
408   return @coalesced_connection.send(request) if @coalesced_connection
409 
410   request.connection = self
411 
412   if @parser && !@write_buffer.full?
413     if @response_received_at && @keep_alive_timeout &&
414        Utils.elapsed_time(@response_received_at) > @keep_alive_timeout
415       # when pushing a request into an existing connection, we have to check whether there
416       # is the possibility that the connection might have extended the keep alive timeout.
417       # for such cases, we want to ping for availability before deciding to shovel requests.
418       log(level: 3) { "keep alive timeout expired, pinging connection..." }
419       @pending << request
420       transition(:active) if @state == :inactive
421       request.ping!
422       ping(request)
423       return
424     end
425 
426     send_request_to_parser(request)
427   else
428     set_connection_timeouts(request) if connecting?
429 
430     @pending << request
431   end
432 end

sibling=(connection)

    # File lib/httpx/connection.rb
478 def sibling=(connection)
479   @sibling = connection
480 
481   return unless connection
482 
483   @main_sibling = connection.sibling.nil?
484 
485   return unless @main_sibling
486 
487   connection.sibling = self
488 end

terminate()

    # File lib/httpx/connection.rb
342 def terminate
343   case @state
344   when :idle
345     purge_after_closed
346 
347     # @fiber-switch-guard
348     if @io.can_disconnect? && @pending.empty?
349       @state = :closed
350       disconnect
351       return
352     end
353   when :closed
354     @connected_at = nil
355   end
356 
357   close
358 end

timeout()

    # File lib/httpx/connection.rb
434 def timeout
435   return if @state == :closed || @state == :inactive
436 
437   timeout = @timeout || @options.timeout.operation_timeout
438 
439   return timeout unless connecting? && (attempt_timeout = connect_attempt_time_left)
440 
441   # the attempt may already be overdue
442   attempt_timeout = 0 unless attempt_timeout.positive?
443 
444   timeout && timeout < attempt_timeout ? timeout : attempt_timeout
445 end

to_io()

    # File lib/httpx/connection.rb
261 def to_io
262   @io.to_io
263 end

used?()

    # File lib/httpx/connection.rb
460 def used?
461   @connected_at
462 end