lib/httpx/connection.rb in httpx-1.3.0 vs lib/httpx/connection.rb in httpx-1.3.1

@@ -560,10 +560,13 @@
 
         @timeout = @current_timeout = parser.timeout
         emit(:open)
       when :inactive
         return unless @state == :open
+
+        # do not deactivate connection in use
+        return if @inflight.positive?
       when :closing
         return unless @state == :idle || @state == :open
 
         unless @write_buffer.empty?
           # preset state before handshake, as error callbacks
@@ -636,11 +639,11 @@
       else
         raise Error, "unsupported transport (#{@type})"
       end
     end
 
-    def on_error(error)
+    def on_error(error, request = nil)
       if error.instance_of?(TimeoutError)
 
         # inactive connections do not contribute to the select loop, therefore
         # they should not fail due to such errors.
         return if @state == :inactive
@@ -650,62 +653,84 @@
           return unless @timeout <= 0
         end
 
         error = error.to_connection_error if connecting?
       end
-      handle_error(error)
+      handle_error(error, request)
       reset
     end
 
-    def handle_error(error)
-      parser.handle_error(error) if @parser && parser.respond_to?(:handle_error)
-      while (request = @pending.shift)
-        response = ErrorResponse.new(request, error)
-        request.response = response
-        request.emit(:response, response)
+    def handle_error(error, request = nil)
+      parser.handle_error(error, request) if @parser && parser.respond_to?(:handle_error)
+      while (req = @pending.shift)
+        next if request && req == request
+
+        response = ErrorResponse.new(req, error)
+        req.response = response
+        req.emit(:response, response)
       end
+
+      return unless request
+
+      response = ErrorResponse.new(request, error)
+      request.response = response
+      request.emit(:response, response)
     end
 
     def set_request_timeouts(request)
-      write_timeout = request.write_timeout
+      set_request_write_timeout(request)
+      set_request_read_timeout(request)
+      set_request_request_timeout(request)
+    end
+
+    def set_request_read_timeout(request)
       read_timeout = request.read_timeout
-      request_timeout = request.request_timeout
 
-      unless write_timeout.nil? || write_timeout.infinite?
-        set_request_timeout(request, write_timeout, :headers, %i[done response]) do
-          write_timeout_callback(request, write_timeout)
-        end
+      return if read_timeout.nil? || read_timeout.infinite?
+
+      set_request_timeout(request, read_timeout, :done, :response) do
+        read_timeout_callback(request, read_timeout)
       end
+    end
 
-      unless read_timeout.nil? || read_timeout.infinite?
-        set_request_timeout(request, read_timeout, :done, :response) do
-          read_timeout_callback(request, read_timeout)
-        end
+    def set_request_write_timeout(request)
+      write_timeout = request.write_timeout
+
+      return if write_timeout.nil? || write_timeout.infinite?
+
+      set_request_timeout(request, write_timeout, :headers, %i[done response]) do
+        write_timeout_callback(request, write_timeout)
       end
+    end
 
+    def set_request_request_timeout(request)
+      request_timeout = request.request_timeout
+
       return if request_timeout.nil? || request_timeout.infinite?
 
-      set_request_timeout(request, request_timeout, :headers, :response) do
+      set_request_timeout(request, request_timeout, :headers, :complete) do
         read_timeout_callback(request, request_timeout, RequestTimeoutError)
       end
     end
 
     def write_timeout_callback(request, write_timeout)
       return if request.state == :done
 
       @write_buffer.clear
       error = WriteTimeoutError.new(request, nil, write_timeout)
-      on_error(error)
+
+      on_error(error, request)
     end
 
     def read_timeout_callback(request, read_timeout, error_type = ReadTimeoutError)
       response = request.response
 
       return if response && response.finished?
 
       @write_buffer.clear
       error = error_type.new(request, request.response, read_timeout)
-      on_error(error)
+
+      on_error(error, request)
     end
 
     def set_request_timeout(request, timeout, start_event, finish_events, &callback)
       request.once(start_event) do
         interval = @timers.after(timeout, callback)