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                                               @family = @io = @ssl_session = @timeout =
51                                                                 @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 
66   if @options.io
67     # if there's an already open IO, get its
68     # peer address, and force-initiate the parser
69     transition(:already_open)
70     @io = build_socket
71     parser
72   else
73     transition(:idle)
74   end
75   self.addresses = @options.addresses if @options.addresses
76 end

Public Instance methods

addresses()

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

addresses?()

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

call()

    # File lib/httpx/connection.rb
214 def call
215   case @state
216   when :idle
217     connect
218 
219     # when opening the tcp or ssl socket fails
220     return if @state == :closed
221 
222     consume
223   when :closed
224     return
225   when :closing
226     consume
227     transition(:closed)
228   when :open
229     consume
230   end
231   nil
232 rescue IOError => e
233   @write_buffer.clear
234   on_io_error(e)
235 rescue StandardError => e
236   @write_buffer.clear
237   on_error(e)
238 rescue Exception => e # rubocop:disable Lint/RescueException
239   force_close(true)
240   raise e
241 end

close()

    # File lib/httpx/connection.rb
243 def close
244   transition(:active) if @state == :inactive
245 
246   @parser.close if @parser
247 end

coalescable?(connection)

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

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

coalesce!(connection)

coalesces self into connection.

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

coalesced?()

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

connecting?()

    # File lib/httpx/connection.rb
181 def connecting?
182   @state == :idle
183 end

deactivate()

    # File lib/httpx/connection.rb
351 def deactivate
352   transition(:inactive)
353 end

disconnect()

disconnects from the current session it’s attached to

    # File lib/httpx/connection.rb
386 def disconnect
387   return if @exhausted # it'll reset
388 
389   return unless (current_session = @current_session) && (current_selector = @current_selector)
390 
391   @current_session = @current_selector = nil
392 
393   current_session.deselect_connection(self, current_selector, @cloned)
394 end

force_close(delete_pending = false)

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

    # File lib/httpx/connection.rb
263 def force_close(delete_pending = false)
264   force_purge
265   return unless @state == :closed
266 
267   if delete_pending
268     @pending.clear
269   elsif (parser = @parser)
270     enqueue_pending_requests_from_parser(parser)
271   end
272 
273   return unless @pending.empty?
274 
275   disconnect
276   emit(:force_closed, delete_pending)
277 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
281 def force_reset(cloned = false)
282   @state = :closing
283   @cloned = cloned
284   transition(:closed)
285 end

handle_connect_error(error)

    # File lib/httpx/connection.rb
377 def handle_connect_error(error)
378   return on_error(error) unless @sibling && @sibling.connecting?
379 
380   @sibling.merge(self)
381 
382   force_reset(true)
383 end

handle_socket_timeout(interval)

    # File lib/httpx/connection.rb
359 def handle_socket_timeout(interval)
360   error = OperationTimeoutError.new(interval, "timed out while waiting on select")
361   error.set_backtrace(caller)
362   on_error(error)
363 end

idling()

    # File lib/httpx/connection.rb
337 def idling
338   purge_after_closed
339   @write_buffer.clear
340   transition(:idle)
341   return unless @parser
342 
343   enqueue_pending_requests_from_parser(parser)
344   @parser = nil
345 end

inflight?()

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

inspect()

:nocov:

    # File lib/httpx/connection.rb
428 def inspect
429   "#<#{self.class}:#{object_id} " \
430     "@origin=#{@origin} " \
431     "@state=#{@state} " \
432     "@pending=#{@pending.size} " \
433     "@io=#{@io}>"
434 end

interests()

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

io_connected?()

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

match?(uri, options)

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

merge(connection)

    # File lib/httpx/connection.rb
148 def merge(connection)
149   @origins |= connection.instance_variable_get(:@origins)
150   if @ssl_session.nil? && connection.ssl_session
151     @ssl_session = connection.ssl_session
152     @io.session_new_cb do |sess|
153       @ssl_session = sess
154     end if @io
155   end
156   connection.purge_pending do |req|
157     req.transition(:idle)
158     send(req)
159   end
160 end

mergeable?(connection)

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

on_connect_error(e)

    # File lib/httpx/connection.rb
396 def on_connect_error(e)
397   # connect errors, exit gracefully
398   error = ConnectionError.new(e.message)
399   error.set_backtrace(e.backtrace)
400   handle_connect_error(error) if connecting?
401   force_close
402 end

on_error(error, request = nil)

    # File lib/httpx/connection.rb
409 def on_error(error, request = nil)
410   if error.is_a?(OperationTimeoutError)
411 
412     # inactive connections do not contribute to the select loop, therefore
413     # they should not fail due to such errors.
414     return if @state == :inactive
415 
416     if @timeout
417       @timeout -= error.timeout
418       return unless @timeout <= 0
419     end
420 
421     error = error.to_connection_error if connecting?
422   end
423   handle_error(error, request)
424   reset
425 end

on_io_error(e)

    # File lib/httpx/connection.rb
404 def on_io_error(e)
405   on_error(e)
406   force_close(true)
407 end

open?()

    # File lib/httpx/connection.rb
355 def open?
356   @state == :open || @state == :inactive
357 end

peer()

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

purge_pending(&block)

    # File lib/httpx/connection.rb
162 def purge_pending(&block)
163   pendings = []
164   if @parser
165     pending = @parser.pending
166     @inflight -= pending.size
167     pendings << pending
168   end
169   pendings << @pending
170   pendings.each do |pending|
171     pending.reject!(&block)
172   end
173 end

reset()

    # File lib/httpx/connection.rb
287 def reset
288   return if @state == :closing || @state == :closed
289 
290   parser = @parser
291 
292   if parser && parser.respond_to?(:max_concurrent_requests)
293     # if connection being reset has at some downgraded the number of concurrent
294     # requests, such as in the case where an attempt to use HTTP/1 pipelining failed,
295     # keep that information around.
296     @max_concurrent_requests = parser.max_concurrent_requests
297   end
298 
299   transition(:closing)
300 
301   transition(:closed)
302 end

send(request)

    # File lib/httpx/connection.rb
304 def send(request)
305   return @coalesced_connection.send(request) if @coalesced_connection
306 
307   if @parser && !@write_buffer.full?
308     if @response_received_at && @keep_alive_timeout &&
309        Utils.elapsed_time(@response_received_at) > @keep_alive_timeout
310       # when pushing a request into an existing connection, we have to check whether there
311       # is the possibility that the connection might have extended the keep alive timeout.
312       # for such cases, we want to ping for availability before deciding to shovel requests.
313       log(level: 3) { "keep alive timeout expired, pinging connection..." }
314       @pending << request
315       transition(:active) if @state == :inactive
316       parser.ping
317       request.ping!
318       return
319     end
320 
321     send_request_to_parser(request)
322   else
323     @pending << request
324   end
325 end

sibling=(connection)

    # File lib/httpx/connection.rb
365 def sibling=(connection)
366   @sibling = connection
367 
368   return unless connection
369 
370   @main_sibling = connection.sibling.nil?
371 
372   return unless @main_sibling
373 
374   connection.sibling = self
375 end

terminate()

    # File lib/httpx/connection.rb
249 def terminate
250   case @state
251   when :idle
252     purge_after_closed
253     disconnect
254   when :closed
255     @connected_at = nil
256   end
257 
258   close
259 end

timeout()

    # File lib/httpx/connection.rb
327 def timeout
328   return if @state == :closed || @state == :inactive
329 
330   return @timeout if @timeout
331 
332   return @options.timeout[:connect_timeout] if @state == :idle
333 
334   @options.timeout[:operation_timeout]
335 end

to_io()

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

used?()

    # File lib/httpx/connection.rb
347 def used?
348   @connected_at
349 end