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
parser_type

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 = @sibling = @coalesced_connection = nil
 54   @exhausted = @cloned = @main_sibling = false
 55 
 56   @options = Options.new(options)
 57   @type = initialize_type(uri, @options)
 58   @origins = [uri.origin]
 59   @origin = Utils.to_uri(uri.origin)
 60   @window_size = @options.window_size
 61   @read_buffer = Buffer.new(@options.buffer_size)
 62   @write_buffer = Buffer.new(@options.buffer_size)
 63   @pending = []
 64 
 65   on(:error, &method(:on_error))
 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   on(:close) do
 76     next if @exhausted # it'll reset
 77 
 78     # may be called after ":close" above, so after the connection has been checked back in.
 79     # next unless @current_session
 80 
 81     next unless @current_session
 82 
 83     @current_session.deselect_connection(self, @current_selector, @cloned)
 84   end
 85   on(:terminate) do
 86     next if @exhausted # it'll reset
 87 
 88     current_session = @current_session
 89     current_selector = @current_selector
 90 
 91     # may be called after ":close" above, so after the connection has been checked back in.
 92     next unless current_session && current_selector
 93 
 94     current_session.deselect_connection(self, current_selector)
 95   end
 96 
 97   on(:altsvc) do |alt_origin, origin, alt_params|
 98     build_altsvc_connection(alt_origin, origin, alt_params)
 99   end
100 
101   @inflight = 0
102   @keep_alive_timeout = @options.timeout[:keep_alive_timeout]
103 
104   @intervals = []
105 
106   self.addresses = @options.addresses if @options.addresses
107 end

parser_type(protocol)

    # File lib/httpx/connection.rb
924 def parser_type(protocol)
925   case protocol
926   when "h2" then HTTP2
927   when "http/1.1" then HTTP1
928   else
929     raise Error, "unsupported protocol (##{protocol})"
930   end
931 end

Public Instance methods

addresses()

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

call()

    # File lib/httpx/connection.rb
241 def call
242   case @state
243   when :idle
244     connect
245     consume
246   when :closed
247     return
248   when :closing
249     consume
250     transition(:closed)
251   when :open
252     consume
253   end
254   nil
255 rescue StandardError => e
256   emit(:error, e)
257   raise e
258 end

close()

    # File lib/httpx/connection.rb
260 def close
261   transition(:active) if @state == :inactive
262 
263   @parser.close if @parser
264 end

coalescable?(connection)

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

    # File lib/httpx/connection.rb
159 def coalescable?(connection)
160   if @io.protocol == "h2" &&
161      @origin.scheme == "https" &&
162      connection.origin.scheme == "https" &&
163      @io.can_verify_peer?
164     @io.verify_hostname(connection.origin.host)
165   else
166     @origin == connection.origin
167   end
168 end

coalesced_connection=(connection)

    # File lib/httpx/connection.rb
353 def coalesced_connection=(connection)
354   @coalesced_connection = connection
355 
356   close_sibling
357   connection.merge(self)
358 end

connecting?()

    # File lib/httpx/connection.rb
205 def connecting?
206   @state == :idle
207 end

create_idle(options = {})

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

deactivate()

    # File lib/httpx/connection.rb
331 def deactivate
332   transition(:inactive)
333 end

expired?()

    # File lib/httpx/connection.rb
140 def expired?
141   return false unless @io
142 
143   @io.expired?
144 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
274 def force_reset(cloned = false)
275   @state = :closing
276   @cloned = cloned
277   transition(:closed)
278 end

handle_connect_error(error)

    # File lib/httpx/connection.rb
372 def handle_connect_error(error)
373   @connect_error = error
374 
375   return handle_error(error) unless @sibling && @sibling.connecting?
376 
377   @sibling.merge(self)
378 
379   force_reset(true)
380 end

handle_socket_timeout(interval)

    # File lib/httpx/connection.rb
339 def handle_socket_timeout(interval)
340   @intervals.delete_if(&:elapsed?)
341 
342   unless @intervals.empty?
343     # remove the intervals which will elapse
344 
345     return
346   end
347 
348   error = HTTPX::TimeoutError.new(interval, "timed out while waiting on select")
349   error.set_backtrace(caller)
350   on_error(error)
351 end

idling()

    # File lib/httpx/connection.rb
320 def idling
321   purge_after_closed
322   @write_buffer.clear
323   transition(:idle)
324   @parser = nil if @parser
325 end

inflight?()

    # File lib/httpx/connection.rb
209 def inflight?
210   @parser && (
211     # parser may be dealing with other requests (possibly started from a different fiber)
212     !@parser.empty? ||
213     # connection may be doing connection termination handshake
214     !@write_buffer.empty?
215   )
216 end

interests()

    # File lib/httpx/connection.rb
218 def interests
219   # connecting
220   if connecting?
221     connect
222 
223     return @io.interests if connecting?
224   end
225 
226   # if the write buffer is full, we drain it
227   return :w unless @write_buffer.empty?
228 
229   return @parser.interests if @parser
230 
231   nil
232 rescue StandardError => e
233   emit(:error, e)
234   nil
235 end

io_connected?()

    # File lib/httpx/connection.rb
199 def io_connected?
200   return @coalesced_connection.io_connected? if @coalesced_connection
201 
202   @io && @io.state == :connected
203 end

match?(uri, options)

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

merge(connection)

    # File lib/httpx/connection.rb
174 def merge(connection)
175   @origins |= connection.instance_variable_get(:@origins)
176   if connection.ssl_session
177     @ssl_session = connection.ssl_session
178     @io.session_new_cb do |sess|
179       @ssl_session = sess
180     end if @io
181   end
182   connection.purge_pending do |req|
183     send(req)
184   end
185 end

mergeable?(connection)

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

open?()

    # File lib/httpx/connection.rb
335 def open?
336   @state == :open || @state == :inactive
337 end

peer()

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

purge_pending(&block)

    # File lib/httpx/connection.rb
187 def purge_pending(&block)
188   pendings = []
189   if @parser
190     @inflight -= @parser.pending.size
191     pendings << @parser.pending
192   end
193   pendings << @pending
194   pendings.each do |pending|
195     pending.reject!(&block)
196   end
197 end

reset()

    # File lib/httpx/connection.rb
280 def reset
281   return if @state == :closing || @state == :closed
282 
283   transition(:closing)
284 
285   transition(:closed)
286 end

send(request)

    # File lib/httpx/connection.rb
288 def send(request)
289   return @coalesced_connection.send(request) if @coalesced_connection
290 
291   if @parser && !@write_buffer.full?
292     if @response_received_at && @keep_alive_timeout &&
293        Utils.elapsed_time(@response_received_at) > @keep_alive_timeout
294       # when pushing a request into an existing connection, we have to check whether there
295       # is the possibility that the connection might have extended the keep alive timeout.
296       # for such cases, we want to ping for availability before deciding to shovel requests.
297       log(level: 3) { "keep alive timeout expired, pinging connection..." }
298       @pending << request
299       transition(:active) if @state == :inactive
300       parser.ping
301       return
302     end
303 
304     send_request_to_parser(request)
305   else
306     @pending << request
307   end
308 end

sibling=(connection)

    # File lib/httpx/connection.rb
360 def sibling=(connection)
361   @sibling = connection
362 
363   return unless connection
364 
365   @main_sibling = connection.sibling.nil?
366 
367   return unless @main_sibling
368 
369   connection.sibling = self
370 end

terminate()

    # File lib/httpx/connection.rb
266 def terminate
267   @connected_at = nil if @state == :closed
268 
269   close
270 end

timeout()

    # File lib/httpx/connection.rb
310 def timeout
311   return if @state == :closed || @state == :inactive
312 
313   return @timeout if @timeout
314 
315   return @options.timeout[:connect_timeout] if @state == :idle
316 
317   @options.timeout[:operation_timeout]
318 end

to_io()

    # File lib/httpx/connection.rb
237 def to_io
238   @io.to_io
239 end

used?()

    # File lib/httpx/connection.rb
327 def used?
328   @connected_at
329 end