http1.rb in httpx-0.12.0

- old
+ new

@@ -67,18 +67,20 @@
         return
       end
 
       return if @requests.include?(request)
 
+      request.once(:headers, &method(:set_protocol_headers))
       @requests << request
       @pipelining = true if @requests.size > 1
     end
 
     def consume
-      requests_limit = [@max_concurrent_requests, @max_requests, @requests.size].min
+      requests_limit = [@max_requests, @requests.size].min
+      concurrent_requests_limit = [@max_concurrent_requests, requests_limit].min
       @requests.each_with_index do |request, idx|
-        break if idx >= requests_limit
+        break if idx >= concurrent_requests_limit
         next if request.state == :done
 
         request.headers["connection"] ||= request.options.persistent || idx < requests_limit - 1 ? "keep-alive" : "close"
         handle(request)
       end
@@ -113,11 +115,11 @@
       return unless @request
 
       response = @request.response
       log(level: 2) { "trailer headers received" }
 
-      log(color: :yellow) { h.each.map { |f, v| "-> HEADER: #{f}: #{v}" }.join("\n") }
+      log(color: :yellow) { h.each.map { |f, v| "-> HEADER: #{f}: #{v.join(", ")}" }.join("\n") }
       response.merge_headers(h)
     end
 
     def on_data(chunk)
       return unless @request
@@ -159,17 +161,17 @@
 
       send(@pending.shift) unless @pending.empty?
     end
 
     def handle_error(ex)
-      if ex.is_a?(EOFError) && @request && @request.response &&
+      if (ex.is_a?(EOFError) || ex.is_a?(TimeoutError)) && @request && @request.response &&
          !@request.response.headers.key?("content-length") &&
          !@request.response.headers.key?("transfer-encoding")
         # if the response does not contain a content-length header, the server closing the
         # connnection is the indicator of response consumed.
         # https://greenbytes.de/tech/webdav/rfc2616.html#rfc.section.4.4
-        on_complete
+        catch(:called) { on_complete }
         return
       end
 
       if @pipelining
         disable
@@ -248,11 +250,11 @@
       # 1 keep alive request.
       @max_concurrent_requests = 1
       @pipelining = false
     end
 
-    def set_request_headers(request)
+    def set_protocol_headers(request)
       request.headers["host"] ||= request.authority
       request.headers["connection"] ||= request.options.persistent ? "keep-alive" : "close"
       if !request.headers.key?("content-length") &&
          request.body.bytesize == Float::INFINITY
         request.chunk!
@@ -262,10 +264,9 @@
     def headline_uri(request)
       request.path
     end
 
     def handle(request)
-      set_request_headers(request)
       catch(:buffer_full) do
         request.transition(:headers)
         join_headers(request) if request.state == :headers
         request.transition(:body)
         join_body(request) if request.state == :body