lib/httpx/connection/http2.rb in httpx-0.16.1 vs lib/httpx/connection/http2.rb in httpx-0.17.0

@@ -1,9 +1,8 @@
 # frozen_string_literal: true
 
 require "securerandom"
-require "io/wait"
 require "http/2/next"
 
 module HTTPX
   class Connection::HTTP2
     include Callbacks
@@ -54,11 +53,11 @@
         return @buffer.empty? ? :r : :rw
       end
 
       return :w if !@pending.empty? && can_buffer_more_requests?
 
-      return :w if @streams.each_key.any? { |r| r.interests == :w }
+      return :w unless @drains.empty?
 
       if @buffer.empty?
         return if @streams.empty? && @pings.empty?
 
         return :r
@@ -216,11 +215,11 @@
     def join_headers(stream, request)
       extra_headers = set_protocol_headers(request)
       log(level: 1, color: :yellow) do
         request.headers.merge(extra_headers).each.map { |k, v| "#{stream.id}: -> HEADER: #{k}: #{v}" }.join("\n")
       end
-      stream.headers(request.headers.each(extra_headers), end_stream: request.empty?)
+      stream.headers(request.headers.each(extra_headers), end_stream: request.body.empty?)
     end
 
     def join_trailers(stream, request)
       unless request.trailers?
         stream.data("", end_stream: true) if request.callbacks_for?(:trailers)
@@ -232,11 +231,11 @@
       end
       stream.headers(request.trailers.each, end_stream: true)
     end
 
     def join_body(stream, request)
-      return if request.empty?
+      return if request.body.empty?
 
       chunk = @drains.delete(request) || request.drain_body
       while chunk
         next_chunk = request.drain_body
         log(level: 1, color: :green) { "#{stream.id}: -> DATA: #{chunk.bytesize} bytes..." }
@@ -247,20 +246,24 @@
           throw(:buffer_full)
         end
         chunk = next_chunk
       end
 
-      on_stream_refuse(stream, request, request.drain_error) if request.drain_error
+      return unless (error = request.drain_error)
+
+      on_stream_refuse(stream, request, error)
     end
 
     ######
     # HTTP/2 Callbacks
     ######
 
     def on_stream_headers(stream, request, h)
-      if request.response && request.response.version == "2.0"
-        on_stream_trailers(stream, request, h)
+      response = request.response
+
+      if response.is_a?(Response) && response.version == "2.0"
+        on_stream_trailers(stream, response, h)
         return
       end
 
       log(color: :yellow) do
         h.map { |k, v| "#{stream.id}: <- HEADER: #{k}: #{v}" }.join("\n")
@@ -272,15 +275,15 @@
       @streams[request] = stream
 
       handle(request, stream) if request.expects?
     end
 
-    def on_stream_trailers(stream, request, h)
+    def on_stream_trailers(stream, response, h)
       log(color: :yellow) do
         h.map { |k, v| "#{stream.id}: <- HEADER: #{k}: #{v}" }.join("\n")
       end
-      request.response.merge_headers(h)
+      response.merge_headers(h)
     end
 
     def on_stream_data(stream, request, data)
       log(level: 1, color: :green) { "#{stream.id}: <- DATA: #{data.bytesize} bytes..." }
       log(level: 2, color: :green) { "#{stream.id}: <- #{data.inspect}" }
@@ -302,10 +305,10 @@
         ex.set_backtrace(caller)
         response = ErrorResponse.new(request, ex, request.options)
         emit(:response, request, response)
       else
         response = request.response
-        if response.status == 421
+        if response && response.status == 421
           ex = MisdirectedRequestError.new(response)
           ex.set_backtrace(caller)
           emit(:error, request, ex)
         else
           emit(:response, request, response)