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

close()

    # File lib/httpx/connection.rb
261 def close
262   transition(:active) if @state == :inactive
263 
264   @parser.close if @parser
265 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
182 def connecting?
183   @state == :idle
184 end

deactivate()

    # File lib/httpx/connection.rb
381 def deactivate
382   transition(:inactive)
383 end

disconnect()

disconnects from the current session it’s attached to

    # File lib/httpx/connection.rb
416 def disconnect
417   return if @exhausted # it'll reset
418 
419   return unless (current_session = @current_session) && (current_selector = @current_selector)
420 
421   @current_session = @current_selector = nil
422 
423   current_session.deselect_connection(self, current_selector, @cloned)
424 end

force_close(delete_pending = false)

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

    # File lib/httpx/connection.rb
286 def force_close(delete_pending = false)
287   force_purge
288   return unless @state == :closed
289 
290   if delete_pending
291     @pending.clear
292   elsif (parser = @parser)
293     enqueue_pending_requests_from_parser(parser)
294   end
295 
296   return unless @pending.empty?
297 
298   disconnect
299   emit(:force_closed, delete_pending)
300 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
304 def force_reset(cloned = false)
305   @state = :closing
306   @cloned = cloned
307   transition(:closed)
308 end

handle_connect_error(error)

    # File lib/httpx/connection.rb
407 def handle_connect_error(error)
408   return on_error(error) unless @sibling && @sibling.connecting?
409 
410   @sibling.merge(self)
411 
412   force_reset(true)
413 end

handle_socket_timeout(interval)

    # File lib/httpx/connection.rb
389 def handle_socket_timeout(interval)
390   error = OperationTimeoutError.new(interval, "timed out while waiting on select")
391   error.set_backtrace(caller)
392   on_error(error)
393 end

idling()

    # File lib/httpx/connection.rb
364 def idling
365   purge_after_closed
366 
367   return unless @state == :closed
368 
369   @write_buffer.clear
370   transition(:idle)
371   return unless @parser
372 
373   enqueue_pending_requests_from_parser(parser)
374   @parser = nil
375 end

inflight?()

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

inspect()

:nocov:

    # File lib/httpx/connection.rb
460 def inspect
461   "#<#{self.class}:#{object_id} " \
462     "@origin=#{@origin} " \
463     "@state=#{@state} " \
464     "@pending=#{@pending.size} " \
465     "@io=#{@io}>"
466 end

interests()

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

io_connected?()

    # File lib/httpx/connection.rb
176 def io_connected?
177   return @coalesced_connection.io_connected? if @coalesced_connection
178 
179   @io && @io.state == :connected
180 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? && connection.ssl_session
152     @ssl_session = connection.ssl_session
153     @io.session_new_cb do |sess|
154       @ssl_session = sess
155     end if @io
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
426 def on_connect_error(e)
427   # connect errors, exit gracefully
428   error = ConnectionError.new(e.message)
429   error.set_backtrace(e.backtrace)
430   handle_connect_error(error) if connecting?
431   force_close
432 end

on_error(error, request = nil)

    # File lib/httpx/connection.rb
439 def on_error(error, request = nil)
440   if error.is_a?(OperationTimeoutError)
441 
442     # inactive connections do not contribute to the select loop, therefore
443     # they should not fail due to such errors.
444     return if @state == :inactive
445 
446     if @timeout
447       @timeout -= error.timeout
448       return unless @timeout <= 0
449 
450       @timeout = nil
451     end
452 
453     error = error.to_connection_error if connecting?
454   end
455   handle_error(error, request)
456   reset
457 end

on_io_error(e)

    # File lib/httpx/connection.rb
434 def on_io_error(e)
435   on_error(e)
436   force_close(true)
437 end

open?()

    # File lib/httpx/connection.rb
385 def open?
386   @state == :open || @state == :inactive
387 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   pendings = []
165   if @parser
166     pending = @parser.pending
167     @inflight -= pending.size
168     pendings << pending
169   end
170   pendings << @pending
171   pendings.each do |pending|
172     pending.reject!(&block)
173   end
174 end

reset()

    # File lib/httpx/connection.rb
310 def reset
311   return if @state == :closing || @state == :closed
312 
313   # do not reset a connection which may have restarted back to :idle, such when the parser resets
314   # (example: HTTP/1 parser disabling pipelining)
315   return if @state == :idle && @pending.any?
316 
317   parser = @parser
318 
319   if parser && parser.respond_to?(:max_concurrent_requests)
320     # if connection being reset has at some downgraded the number of concurrent
321     # requests, such as in the case where an attempt to use HTTP/1 pipelining failed,
322     # keep that information around.
323     @max_concurrent_requests = parser.max_concurrent_requests
324   end
325 
326   transition(:closing)
327 
328   transition(:closed)
329 end

send(request)

    # File lib/httpx/connection.rb
331 def send(request)
332   return @coalesced_connection.send(request) if @coalesced_connection
333 
334   if @parser && !@write_buffer.full?
335     if @response_received_at && @keep_alive_timeout &&
336        Utils.elapsed_time(@response_received_at) > @keep_alive_timeout
337       # when pushing a request into an existing connection, we have to check whether there
338       # is the possibility that the connection might have extended the keep alive timeout.
339       # for such cases, we want to ping for availability before deciding to shovel requests.
340       log(level: 3) { "keep alive timeout expired, pinging connection..." }
341       @pending << request
342       transition(:active) if @state == :inactive
343       request.ping!
344       ping(request)
345       return
346     end
347 
348     send_request_to_parser(request)
349   else
350     @pending << request
351   end
352 end

sibling=(connection)

    # File lib/httpx/connection.rb
395 def sibling=(connection)
396   @sibling = connection
397 
398   return unless connection
399 
400   @main_sibling = connection.sibling.nil?
401 
402   return unless @main_sibling
403 
404   connection.sibling = self
405 end

terminate()

    # File lib/httpx/connection.rb
267 def terminate
268   case @state
269   when :idle
270     purge_after_closed
271 
272     # @fiber-switch-guard
273     if @io.can_disconnect? && @pending.empty?
274       disconnect
275       return
276     end
277   when :closed
278     @connected_at = nil
279   end
280 
281   close
282 end

timeout()

    # File lib/httpx/connection.rb
354 def timeout
355   return if @state == :closed || @state == :inactive
356 
357   return @timeout if @timeout
358 
359   return @options.timeout[:connect_timeout] if @state == :idle
360 
361   @options.timeout[:operation_timeout]
362 end

to_io()

    # File lib/httpx/connection.rb
211 def to_io
212   @io.to_io
213 end

used?()

    # File lib/httpx/connection.rb
377 def used?
378   @connected_at
379 end