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

sibling
ssl_session

Included modules

Classes and Modules

Attributes

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

Public Class methods

new(uri, options)

   # File lib/httpx/connection.rb
47 def initialize(uri, options)
48   @current_session = @current_selector = @max_concurrent_requests =
49                        @parser = @sibling = @coalesced_connection = @altsvc_connection =
50                                               @ping_timer = @family = @io = @ssl_session =
51                                                                         @timeout = @connected_at = @response_received_at = nil
52 
53   @exhausted = @cloned = @main_sibling = false
54 
55   @options = Options.new(options)
56   @type = initialize_type(uri, @options)
57   @origins = [uri.origin]
58   @origin = Utils.to_uri(uri.origin)
59   @window_size = @options.window_size
60   @read_buffer = Buffer.new(@options.buffer_size)
61   @write_buffer = Buffer.new(@options.buffer_size)
62   @pending = []
63   @inflight = 0
64   @keep_alive_timeout = @options.timeout[:keep_alive_timeout]
65   @no_more_requests_counter = 0
66 
67   if @options.io
68     # if there's an already open IO, get its
69     # peer address, and force-initiate the parser
70     transition(:already_open)
71     @io = build_socket
72     parser
73   else
74     transition(:idle)
75   end
76   self.addresses = @options.addresses if @options.addresses
77 end

Public Instance methods

addresses()

   # File lib/httpx/connection.rb
93 def addresses
94   @io && @io.addresses
95 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
85 def addresses=(addrs)
86   if @io
87     @io.add_addresses(addrs)
88   else
89     @io = build_socket(addrs)
90   end
91 end

addresses?()

   # File lib/httpx/connection.rb
97 def addresses?
98   @io && @io.addresses?
99 end

call()

    # File lib/httpx/connection.rb
217 def call
218   case @state
219   when :idle
220     return if no_more_requests?
221 
222     connect
223 
224     # when opening the tcp or ssl socket fails
225     return if @state == :closed
226 
227     consume
228   when :closed
229     return if no_more_requests?
230 
231     # there are pending requests to send, restart the state machine.
232     idling
233 
234     # @fiber-switch-guard
235     # fiber may have switch after ensuring that @io is closed.
236     return unless @state == :idle
237 
238     call
239   when :closing
240     consume
241     transition(:closed)
242 
243     # @fiber-switch-guard
244     # fiber may have switch while closing @io.
245     return if @state == :closed &&
246               # only remain here if there are pending requests.
247               @pending.empty?
248 
249     call
250   when :open
251     consume
252   end
253   nil
254 rescue Errno::ECONNRESET,
255        Errno::EINVAL,
256        SocketError,
257        IOError,
258        TLSError => e
259   @write_buffer.clear
260   on_io_error(e)
261 rescue Error => e
262   @write_buffer.clear
263   on_error(e)
264 rescue Exception => e # rubocop:disable Lint/RescueException
265   force_close(true)
266   raise e
267 end

close()

    # File lib/httpx/connection.rb
273 def close
274   transition(:active) if @state == :inactive
275 
276   @parser.close if @parser
277 end

coalescable?(connection)

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

    # File lib/httpx/connection.rb
138 def coalescable?(connection)
139   if @io.protocol == "h2" &&
140      @origin.scheme == "https" &&
141      connection.origin.scheme == "https" &&
142      @io.can_verify_peer?
143     @io.verify_hostname(connection.origin.host)
144   else
145     @origin == connection.origin
146   end
147 end

coalesce!(connection)

coalesces self into connection.

    # File lib/httpx/connection.rb
125 def coalesce!(connection)
126   @coalesced_connection = connection
127 
128   close_sibling
129   connection.merge(self)
130 end

coalesced?()

    # File lib/httpx/connection.rb
132 def coalesced?
133   @coalesced_connection
134 end

connecting?()

    # File lib/httpx/connection.rb
184 def connecting?
185   @state == :idle
186 end

deactivate()

    # File lib/httpx/connection.rb
398 def deactivate
399   transition(:inactive)
400 end

disconnect()

disconnects from the current session it’s attached to

    # File lib/httpx/connection.rb
433 def disconnect
434   return if @exhausted # it'll reset
435 
436   return unless (current_session = @current_session) && (current_selector = @current_selector)
437 
438   @current_session = @current_selector = nil
439 
440   current_session.deselect_connection(self, current_selector, @cloned)
441 end

force_close(delete_pending = false)

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

    # File lib/httpx/connection.rb
298 def force_close(delete_pending = false)
299   force_purge
300   return unless @state == :closed
301 
302   if delete_pending
303     @pending.clear
304   elsif (parser = @parser)
305     enqueue_pending_requests_from_parser(parser)
306   end
307 
308   return unless @pending.empty?
309 
310   disconnect
311   emit(:force_closed, delete_pending)
312 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
316 def force_reset(cloned = false)
317   @state = :closing
318   @cloned = cloned
319   transition(:closed)
320 end

handle_connect_error(error)

    # File lib/httpx/connection.rb
424 def handle_connect_error(error)
425   return on_error(error) unless @sibling && @sibling.connecting?
426 
427   @sibling.merge(self)
428 
429   force_reset(true)
430 end

handle_socket_timeout(interval)

    # File lib/httpx/connection.rb
406 def handle_socket_timeout(interval)
407   error = OperationTimeoutError.new(interval, "timed out while waiting on select")
408   error.set_backtrace(caller)
409   on_error(error)
410 end

idling()

    # File lib/httpx/connection.rb
381 def idling
382   purge_after_closed
383 
384   return unless @state == :closed
385 
386   @write_buffer.clear
387   transition(:idle)
388   return unless @parser
389 
390   enqueue_pending_requests_from_parser(parser)
391   @parser = nil
392 end

inflight?()

    # File lib/httpx/connection.rb
188 def inflight?
189   @parser && (
190     # parser may be dealing with other requests (possibly started from a different fiber)
191     !@parser.empty? ||
192     # connection may be doing connection termination handshake
193     !@write_buffer.empty?
194   )
195 end

initial_call()

    # File lib/httpx/connection.rb
269 def initial_call
270   call
271 end

inspect()

simplecov:disable

    # File lib/httpx/connection.rb
482 def inspect
483   "#<#{self.class}:#{object_id} " \
484     "@origin=#{@origin} " \
485     "@state=#{@state} " \
486     "@pending=#{@pending.size} " \
487     "@io=#{@io}>"
488 end

interests()

    # File lib/httpx/connection.rb
197 def interests
198   # connecting
199   if connecting?
200     connect
201 
202     return @io.interests if connecting?
203   end
204 
205   return @parser.interests if @parser
206 
207   nil
208 rescue Error => e
209   on_error(e)
210   nil
211 end

io_connected?()

    # File lib/httpx/connection.rb
178 def io_connected?
179   return @coalesced_connection.io_connected? if @coalesced_connection
180 
181   @io && @io.state == :connected
182 end

match?(uri, options)

    # File lib/httpx/connection.rb
101 def match?(uri, options)
102   return false if !used? && (@state == :closing || @state == :closed)
103 
104   @origins.include?(uri.origin) &&
105     # if there is more than one origin to match, it means that this connection
106     # was the result of coalescing. To prevent blind trust in the case where the
107     # origin came from an ORIGIN frame, we're going to verify the hostname with the
108     # SSL certificate
109     (@origins.size == 1 || @origin == uri.origin || (@io.is_a?(SSL) && @io.verify_hostname(uri.host))) &&
110     @options.connection_options_match?(options)
111 end

merge(connection)

    # File lib/httpx/connection.rb
149 def merge(connection)
150   @origins |= connection.instance_variable_get(:@origins)
151   if @ssl_session.nil? && (ssl_session = connection.ssl_session)
152     @ssl_session = ssl_session
153     # the socket only needs the merged session if it can still resume it,
154     # i.e. if TLS hasn't been negotiated yet.
155     @io.ssl_session = ssl_session if @io.is_a?(SSL) && !@io.connected?
156   end
157   connection.purge_pending do |req|
158     req.transition(:idle)
159     send(req)
160   end
161 end

mergeable?(connection)

    # File lib/httpx/connection.rb
113 def mergeable?(connection)
114   return false if @state == :closing || @state == :closed || !@io
115 
116   return false unless connection.addresses
117 
118   (
119     (open? && @origin == connection.origin) ||
120     !(@io.addresses & (connection.addresses || [])).empty?
121   ) && @options.connection_options_match?(connection.options)
122 end

on_connect_error(e)

    # File lib/httpx/connection.rb
443 def on_connect_error(e)
444   # connect errors, exit gracefully
445   error = ConnectionError.new(e.message)
446   error.set_backtrace(e.backtrace)
447   handle_connect_error(error) if connecting?
448   force_close
449 end

on_error(error, request = nil)

    # File lib/httpx/connection.rb
461 def on_error(error, request = nil)
462   if error.is_a?(OperationTimeoutError)
463 
464     # inactive connections do not contribute to the select loop, therefore
465     # they should not fail due to such errors.
466     return if @state == :inactive
467 
468     if @timeout
469       @timeout -= error.timeout
470       return unless @timeout <= 0
471 
472       @timeout = nil
473     end
474 
475     error = error.to_connection_error if connecting?
476   end
477   handle_error(error, request)
478   reset
479 end

on_io_error(e)

    # File lib/httpx/connection.rb
451 def on_io_error(e)
452   on_error(e)
453 
454   # do not force close if parser resets the connection.
455   # can happen i.e. when HTTP/1.1 pipelining is disabled.
456   return if @state == :idle && @pending.any?
457 
458   force_close(true)
459 end

open?()

    # File lib/httpx/connection.rb
402 def open?
403   @state == :open || @state == :inactive
404 end

peer()

   # File lib/httpx/connection.rb
79 def peer
80   @origin
81 end

purge_pending(&block)

    # File lib/httpx/connection.rb
163 def purge_pending(&block)
164   if @parser
165     pending = @parser.pending
166     @inflight -= pending.size
167     pending.reject! do |req|
168       block.call(req)
169       true
170     end
171   end
172   @pending.reject! do |req|
173     block.call(req)
174     true
175   end
176 end

reset()

    # File lib/httpx/connection.rb
322 def reset
323   return if @state == :closing || @state == :closed
324 
325   # do not reset a connection which may have restarted back to :idle, such when the parser resets
326   # (example: HTTP/1 parser disabling pipelining)
327   return if @state == :idle && @pending.any?
328 
329   if @ping_timer
330     @ping_timer.cancel
331     @ping_timer = nil
332   end
333 
334   parser = @parser
335 
336   if parser && parser.respond_to?(:max_concurrent_requests)
337     # if connection being reset has at some downgraded the number of concurrent
338     # requests, such as in the case where an attempt to use HTTP/1 pipelining failed,
339     # keep that information around.
340     @max_concurrent_requests = parser.max_concurrent_requests
341   end
342 
343   transition(:closing)
344 
345   transition(:closed)
346 end

send(request)

    # File lib/httpx/connection.rb
348 def send(request)
349   return @coalesced_connection.send(request) if @coalesced_connection
350 
351   if @parser && !@write_buffer.full?
352     if @response_received_at && @keep_alive_timeout &&
353        Utils.elapsed_time(@response_received_at) > @keep_alive_timeout
354       # when pushing a request into an existing connection, we have to check whether there
355       # is the possibility that the connection might have extended the keep alive timeout.
356       # for such cases, we want to ping for availability before deciding to shovel requests.
357       log(level: 3) { "keep alive timeout expired, pinging connection..." }
358       @pending << request
359       transition(:active) if @state == :inactive
360       request.ping!
361       ping(request)
362       return
363     end
364 
365     send_request_to_parser(request)
366   else
367     @pending << request
368   end
369 end

sibling=(connection)

    # File lib/httpx/connection.rb
412 def sibling=(connection)
413   @sibling = connection
414 
415   return unless connection
416 
417   @main_sibling = connection.sibling.nil?
418 
419   return unless @main_sibling
420 
421   connection.sibling = self
422 end

terminate()

    # File lib/httpx/connection.rb
279 def terminate
280   case @state
281   when :idle
282     purge_after_closed
283 
284     # @fiber-switch-guard
285     if @io.can_disconnect? && @pending.empty?
286       disconnect
287       return
288     end
289   when :closed
290     @connected_at = nil
291   end
292 
293   close
294 end

timeout()

    # File lib/httpx/connection.rb
371 def timeout
372   return if @state == :closed || @state == :inactive
373 
374   return @timeout if @timeout
375 
376   return @options.timeout[:connect_timeout] if @state == :idle
377 
378   @options.timeout[:operation_timeout]
379 end

to_io()

    # File lib/httpx/connection.rb
213 def to_io
214   @io.to_io
215 end

used?()

    # File lib/httpx/connection.rb
394 def used?
395   @connected_at
396 end