connection.rb in httpx-0.18.0

- old
+ new

@@ -68,11 +68,11 @@
         transition(:idle)
       end
 
       @inflight = 0
       @keep_alive_timeout = @options.timeout[:keep_alive_timeout]
-      @keep_alive_timer = nil
+      @total_timeout = @options.timeout[:total_timeout]
 
       self.addresses = @options.addresses if @options.addresses
     end
 
     # this is a semi-private method, to be used by the resolver
@@ -198,15 +198,13 @@
       end
       nil
     end
 
     def close
-      @parser.close if @parser
-      return unless @keep_alive_timer
+      transition(:active) if @state == :inactive
 
-      @keep_alive_timer.cancel
-      remove_instance_variable(:@keep_alive_timer)
+      @parser.close if @parser
     end
 
     def reset
       transition(:closing)
       transition(:closed)
@@ -214,37 +212,59 @@
     end
 
     def send(request)
       if @parser && !@write_buffer.full?
         request.headers["alt-used"] = @origin.authority if match_altsvcs?(request.uri)
-        if @keep_alive_timer
+
+        if @response_received_at && @keep_alive_timeout &&
+           Utils.elapsed_time(@response_received_at) > @keep_alive_timeout
           # when pushing a request into an existing connection, we have to check whether there
           # is the possibility that the connection might have extended the keep alive timeout.
           # for such cases, we want to ping for availability before deciding to shovel requests.
-          if @keep_alive_timer.fires_in.negative?
-            @pending << request
-            parser.ping
-            return
-          end
-
-          @keep_alive_timer.pause
+          @pending << request
+          parser.ping
+          transition(:active) if @state == :inactive
+          return
         end
-        @inflight += 1
-        parser.send(request)
+
+        send_request_to_parser(request)
       else
         @pending << request
       end
     end
 
     def timeout
+      if @total_timeout
+        return @total_timeout unless @connected_at
+
+        elapsed_time = @total_timeout - Utils.elapsed_time(@connected_at)
+
+        if elapsed_time.negative?
+          ex = TotalTimeoutError.new(@total_timeout, "Timed out after #{@total_timeout} seconds")
+          ex.set_backtrace(caller)
+          on_error(@total_timeout)
+          return
+        end
+
+        return elapsed_time
+      end
+
       return @timeout if defined?(@timeout)
 
       return @options.timeout[:connect_timeout] if @state == :idle
 
       @options.timeout[:operation_timeout]
     end
 
+    def deactivate
+      transition(:inactive)
+    end
+
+    def open?
+      @state == :open || @state == :inactive
+    end
+
     private
 
     def connect
       transition(:open)
     end
@@ -377,20 +397,27 @@
       end
     end
 
     def send_pending
       while !@write_buffer.full? && (request = @pending.shift)
-        @inflight += 1
-        @keep_alive_timer.pause if @keep_alive_timer
-        parser.send(request)
+        send_request_to_parser(request)
       end
     end
 
     def parser
       @parser ||= build_parser
     end
 
+    def send_request_to_parser(request)
+      @inflight += 1
+      parser.send(request)
+
+      return unless @state == :inactive
+
+      transition(:active)
+    end
+
     def build_parser(protocol = @io.protocol)
       parser = registry(protocol).new(@write_buffer, @options)
       set_parser_callbacks(parser)
       parser
     end
@@ -398,11 +425,12 @@
     def set_parser_callbacks(parser)
       parser.on(:response) do |request, response|
         AltSvc.emit(request, response) do |alt_origin, origin, alt_params|
           emit(:altsvc, alt_origin, origin, alt_params)
         end
-        handle_response
+        @response_received_at = Utils.now
+        @inflight -= 1
         request.emit(:response, response)
       end
       parser.on(:altsvc) do |alt_origin, origin, alt_params|
         emit(:altsvc, alt_origin, origin, alt_params)
       end
@@ -418,11 +446,11 @@
       parser.on(:origin) do |origin|
         @origins |= [origin]
       end
       parser.on(:close) do |force|
         transition(:closing)
-        if force
+        if force || @state == :idle
           transition(:closed)
           emit(:close)
         end
       end
       parser.on(:close_handshake) do
@@ -464,35 +492,37 @@
         @timeout = @current_timeout = @options.timeout[:connect_timeout]
 
       when :open
         return if @state == :closed
 
-        total_timeout
-
         @io.connect
         return unless @io.connected?
 
+        @connected_at = Utils.now
+
         send_pending
 
         @timeout = @current_timeout = parser.timeout
         emit(:open)
+      when :inactive
+        return unless @state == :open
       when :closing
         return unless @state == :open
 
       when :closed
         return unless @state == :closing
         return unless @write_buffer.empty?
 
-        if @total_timeout
-          @total_timeout.cancel
-          remove_instance_variable(:@total_timeout)
-        end
-
         purge_after_closed
       when :already_open
         nextstate = :open
         send_pending
+      when :active
+        return unless @state == :inactive
+
+        nextstate = :open
+        emit(:activate)
       end
       @state = nextstate
     rescue Errno::ECONNREFUSED,
            Errno::EADDRNOTAVAIL,
            Errno::EHOSTUNREACH,
@@ -504,48 +534,28 @@
     end
 
     def purge_after_closed
       @io.close if @io
       @read_buffer.clear
-      if @keep_alive_timer
-        @keep_alive_timer.cancel
-        remove_instance_variable(:@keep_alive_timer)
-      end
-
       remove_instance_variable(:@timeout) if defined?(@timeout)
     end
 
-    def handle_response
-      @inflight -= 1
-      return unless @inflight.zero?
-
-      if @keep_alive_timer
-        @keep_alive_timer.resume
-        @keep_alive_timer.reset
-      else
-        @keep_alive_timer = @timers.after(@keep_alive_timeout) do
-          unless @inflight.zero?
-            log { "(#{@origin}): keep alive timeout expired" }
-            parser.ping
-          end
-        end
-      end
-    end
-
     def on_error(error)
       if error.instance_of?(TimeoutError)
-        if @timeout
-          @timeout -= error.timeout
-          return unless @timeout <= 0
-        end
 
-        if @total_timeout && @total_timeout.fires_in.negative?
-          ex = TotalTimeoutError.new(@total_timeout.interval, "Timed out after #{@total_timeout.interval} seconds")
+        if @total_timeout && @connected_at &&
+           Utils.elapsed_time(@connected_at) > @total_timeout
+          ex = TotalTimeoutError.new(@total_timeout, "Timed out after #{@total_timeout} seconds")
           ex.set_backtrace(error.backtrace)
           error = ex
-        elsif connecting?
-          error = error.to_connection_error
+        else
+          if @timeout
+            @timeout -= error.timeout
+            return unless @timeout <= 0
+          end
+
+          error = error.to_connection_error if connecting?
         end
       end
       handle_error(error)
       reset
     end
@@ -554,22 +564,9 @@
       parser.handle_error(error) if @parser && parser.respond_to?(:handle_error)
       while (request = @pending.shift)
         response = ErrorResponse.new(request, error, @options)
         request.response = response
         request.emit(:response, response)
-      end
-    end
-
-    def total_timeout
-      total = @options.timeout[:total_timeout]
-
-      return unless total
-
-      @total_timeout ||= @timers.after(total) do
-        ex = TotalTimeoutError.new(total, "Timed out after #{total} seconds")
-        ex.set_backtrace(caller)
-        on_error(ex)
-        @parser.close if @parser
       end
     end
   end
 end