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

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
 52 def initialize(uri, options)
 53   @current_session = @current_selector =
 54     @parser = @sibling = @coalesced_connection =
 55                 @io = @ssl_session = @timeout =
 56                         @connected_at = @response_received_at = nil
 57 
 58   @exhausted = @cloned = @main_sibling = 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 
 71   on(:error, &method(:on_error))
 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   on(:close) do
 82     next if @exhausted # it'll reset
 83 
 84     # may be called after ":close" above, so after the connection has been checked back in.
 85     # next unless @current_session
 86 
 87     next unless @current_session
 88 
 89     @current_session.deselect_connection(self, @current_selector, @cloned)
 90   end
 91   on(:terminate) do
 92     next if @exhausted # it'll reset
 93 
 94     current_session = @current_session
 95     current_selector = @current_selector
 96 
 97     # may be called after ":close" above, so after the connection has been checked back in.
 98     next unless current_session && current_selector
 99 
100     current_session.deselect_connection(self, current_selector)
101   end
102 
103   on(:altsvc) do |alt_origin, origin, alt_params|
104     build_altsvc_connection(alt_origin, origin, alt_params)
105   end
106 
107   self.addresses = @options.addresses if @options.addresses
108 end

Public Instance methods

addresses()

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

call()

    # File lib/httpx/connection.rb
255 def call
256   case @state
257   when :idle
258     connect
259     consume
260   when :closed
261     return
262   when :closing
263     consume
264     transition(:closed)
265   when :open
266     consume
267   end
268   nil
269 rescue StandardError => e
270   @write_buffer.clear
271   emit(:error, e)
272   raise e
273 end

close()

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

coalescable?(connection)

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

    # File lib/httpx/connection.rb
168 def coalescable?(connection)
169   if @io.protocol == "h2" &&
170      @origin.scheme == "https" &&
171      connection.origin.scheme == "https" &&
172      @io.can_verify_peer?
173     @io.verify_hostname(connection.origin.host)
174   else
175     @origin == connection.origin
176   end
177 end

coalesce!(connection)

coalesces self into connection.

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

connecting?()

    # File lib/httpx/connection.rb
220 def connecting?
221   @state == :idle
222 end

create_idle(options = {})

    # File lib/httpx/connection.rb
179 def create_idle(options = {})
180   self.class.new(@origin, @options.merge(options))
181 end

current_context?()

    # File lib/httpx/connection.rb
208 def current_context?
209   @pending.any?(&:current_context?) || (
210     @sibling && @sibling.pending.any?(&:current_context?)
211   )
212 end

deactivate()

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

disconnect()

    # File lib/httpx/connection.rb
387 def disconnect
388   return unless @current_session && @current_selector
389 
390   emit(:close)
391   @current_session = @current_selector = nil
392 end

expired?()

    # File lib/httpx/connection.rb
141 def expired?
142   return false unless @io
143 
144   @io.expired?
145 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
295 def force_reset(cloned = false)
296   @state = :closing
297   @cloned = cloned
298   transition(:closed)
299 end

handle_connect_error(error)

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

handle_socket_timeout(interval)

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

idling()

    # File lib/httpx/connection.rb
342 def idling
343   purge_after_closed
344   @write_buffer.clear
345   transition(:idle)
346   @parser = nil if @parser
347 end

inflight?()

    # File lib/httpx/connection.rb
224 def inflight?
225   @parser && (
226     # parser may be dealing with other requests (possibly started from a different fiber)
227     !@parser.empty? ||
228     # connection may be doing connection termination handshake
229     !@write_buffer.empty?
230   )
231 end

inspect()

:nocov:

    # File lib/httpx/connection.rb
395 def inspect
396   "#<#{self.class}:#{object_id} " \
397     "@origin=#{@origin} " \
398     "@state=#{@state} " \
399     "@pending=#{@pending.size} " \
400     "@io=#{@io}>"
401 end

interests()

    # File lib/httpx/connection.rb
233 def interests
234   # connecting
235   if connecting?
236     return unless @pending.any?(&:current_context?)
237 
238     connect
239 
240     return @io.interests if connecting?
241   end
242 
243   return @parser.interests if @parser
244 
245   nil
246 rescue StandardError => e
247   emit(:error, e)
248   nil
249 end

io_connected?()

    # File lib/httpx/connection.rb
214 def io_connected?
215   return @coalesced_connection.io_connected? if @coalesced_connection
216 
217   @io && @io.state == :connected
218 end

match?(uri, options)

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

merge(connection)

    # File lib/httpx/connection.rb
183 def merge(connection)
184   @origins |= connection.instance_variable_get(:@origins)
185   if connection.ssl_session
186     @ssl_session = connection.ssl_session
187     @io.session_new_cb do |sess|
188       @ssl_session = sess
189     end if @io
190   end
191   connection.purge_pending do |req|
192     send(req)
193   end
194 end

mergeable?(connection)

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

open?()

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

peer()

    # File lib/httpx/connection.rb
110 def peer
111   @origin
112 end

purge_pending(&block)

    # File lib/httpx/connection.rb
196 def purge_pending(&block)
197   pendings = []
198   if @parser
199     @inflight -= @parser.pending.size
200     pendings << @parser.pending
201   end
202   pendings << @pending
203   pendings.each do |pending|
204     pending.reject!(&block)
205   end
206 end

reset()

    # File lib/httpx/connection.rb
301 def reset
302   return if @state == :closing || @state == :closed
303 
304   transition(:closing)
305 
306   transition(:closed)
307 end

send(request)

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

sibling=(connection)

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

terminate()

    # File lib/httpx/connection.rb
281 def terminate
282   case @state
283   when :idle
284     purge_after_closed
285     emit(:terminate)
286   when :closed
287     @connected_at = nil
288   end
289 
290   close
291 end

timeout()

    # File lib/httpx/connection.rb
332 def timeout
333   return if @state == :closed || @state == :inactive
334 
335   return @timeout if @timeout
336 
337   return @options.timeout[:connect_timeout] if @state == :idle
338 
339   @options.timeout[:operation_timeout]
340 end

to_io()

    # File lib/httpx/connection.rb
251 def to_io
252   @io.to_io
253 end

used?()

    # File lib/httpx/connection.rb
349 def used?
350   @connected_at
351 end