Implement IO#readpartial API for Connection, Body, and Inflater

Connection#readpartial now raises EOFError instead of returning nil at
end-of-stream and accepts an optional outbuf parameter, conforming to
Ruby's IO#readpartial contract. This enables IO.copy_stream and other
IO-expecting interfaces to work directly with connection objects.

Closes #618.

Author: sferik

CHANGELOG.md

@@ -18,6 +18,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **BREAKING** `Connection#readpartial` now raises `EOFError` instead of
+  returning `nil` at end-of-stream, and supports an `outbuf` parameter,
+  conforming to the `IO#readpartial` API. `Body#readpartial` and
+  `Inflater#readpartial` also raise `EOFError` (#618)
 - **BREAKING** Stricter timeout options parsing: `.timeout()` with a Hash now
   rejects unknown keys, non-numeric values, string keys, and empty hashes (#754)
 - Bumped min llhttp dependency version

lib/http/connection.rb

@@ -104,23 +104,24 @@ def send_request(req)
     # @example
     #   connection.readpartial
     #
+    # @param [Integer] size maximum bytes to read
+    # @param [String, nil] outbuf buffer to fill with data
     # @return [String] data chunk
-    # @return [nil] when no more data left
+    # @raise [EOFError] when no more data left
     # @api public
-    def readpartial(size = BUFFER_SIZE)
-      return unless @pending_response
+    def readpartial(size = BUFFER_SIZE, outbuf = nil)
+      raise EOFError unless @pending_response
 
       chunk = @parser.read(size)
-      return chunk if chunk
-
-      eof = read_more(size) == :eof
-      check_premature_eof(eof)
-
-      finished = eof || @parser.finished?
-      chunk    = @parser.read(size)
-      finish_response if finished
+      unless chunk
+        eof = read_more(size) == :eof
+        check_premature_eof(eof)
+        finished = eof || @parser.finished?
+        chunk    = @parser.read(size) || "".b
+        finish_response if finished
+      end
 
-      chunk || "".b
+      outbuf ? outbuf.replace(chunk) : chunk
     end
 
     # Reads data from socket up until headers are loaded
@@ -233,18 +234,6 @@ def check_premature_eof(eof)
       raise ConnectionError, "response body ended prematurely"
     end
 
-    # Check if the response body has a known framing mechanism
-    #
-    # @example
-    #   body_framed?
-    #
-    # @return [Boolean]
-    # @api private
-    def body_framed?
-      @parser.headers.include?(Headers::TRANSFER_ENCODING) ||
-        @parser.headers.include?(Headers::CONTENT_LENGTH)
-    end
-
     # Connect socket and set up proxy/TLS
     # @return [void]
     # @api private

lib/http/connection/internals.rb

@@ -79,6 +79,18 @@ def set_keep_alive
           end
       end
 
+      # Check if the response body has a known framing mechanism
+      #
+      # @example
+      #   body_framed?
+      #
+      # @return [Boolean]
+      # @api private
+      def body_framed?
+        @parser.headers.include?(Headers::TRANSFER_ENCODING) ||
+          @parser.headers.include?(Headers::CONTENT_LENGTH)
+      end
+
       # Feeds some more data into parser
       # @return [void]
       # @raise [SocketReadError] when unable to read from socket

lib/http/response.rb

@@ -117,7 +117,8 @@ def initialize(opts)
     #   Read a chunk of the response body
     #   @example
     #     response.readpartial # => "chunk"
-    #   @return [String, nil]
+    #   @return [String]
+    #   @raise [EOFError] when no more data left
     #   @api public
     def_delegator :@body, :readpartial
 

lib/http/response/body.rb

@@ -44,13 +44,13 @@ def initialize(stream, encoding: Encoding::BINARY)
       #   body.readpartial # => "chunk of data"
       #
       # (see HTTP::Client#readpartial)
-      # @return [String, nil]
+      # @return [String]
+      # @raise [EOFError] when no more data left
       # @api public
       def readpartial(*)
         stream!
         chunk = @stream.readpartial(*)
-
-        String.new(chunk, encoding: @encoding) if chunk
+        String.new(chunk, encoding: @encoding)
       end
 
       # Iterate over the body, allowing it to be enumerable
@@ -63,9 +63,10 @@ def readpartial(*)
       # @return [void]
       # @api public
       def each
-        while (chunk = readpartial)
-          yield chunk
+        loop do
+          yield readpartial
         end
+      rescue EOFError # rubocop:disable Lint/SuppressedException
       end
 
       # Eagerly consume the entire body as a string
@@ -125,9 +126,10 @@ def inspect
       def read_contents
         contents = String.new("", encoding: @encoding)
 
-        while (chunk = @stream.readpartial)
-          contents << String.new(chunk, encoding: @encoding)
-          chunk = nil # deallocate string
+        loop do
+          contents << String.new(@stream.readpartial, encoding: @encoding)
+        rescue EOFError
+          break
         end
 
         contents

lib/http/response/inflater.rb

@@ -32,18 +32,19 @@ def initialize(connection)
       # @example
       #   inflater.readpartial # => "decompressed data"
       #
-      # @return [String, nil]
+      # @return [String]
+      # @raise [EOFError] when no more data left
       # @api public
       def readpartial(*)
         chunk = @connection.readpartial(*)
-        return zstream.inflate(chunk) if chunk
-
+        zstream.inflate(chunk)
+      rescue EOFError
         unless zstream.closed?
           zstream.finish if zstream.total_in.positive?
           zstream.close
         end
 
-        nil
+        raise
       end
 
       private

sig/http.rbs

@@ -659,7 +659,7 @@ module HTTP
     def code: () -> Integer
     def to_s: () -> String
     alias to_str to_s
-    def readpartial: (*untyped args) -> String?
+    def readpartial: (*untyped args) -> String
     def connection: () -> untyped
     def uri: () -> URI
     def to_a: () -> Array[untyped]
@@ -724,7 +724,7 @@ module HTTP
       attr_reader connection: untyped
 
       def initialize: (untyped stream, ?encoding: untyped) -> void
-      def readpartial: (*untyped args) -> untyped
+      def readpartial: (*untyped args) -> String
       def each: () { (String) -> void } -> void
              | () -> Enumerator[String, void]
       def to_s: () -> untyped
@@ -744,7 +744,7 @@ module HTTP
       attr_reader connection: untyped
 
       def initialize: (untyped connection) -> void
-      def readpartial: (*untyped args) -> String?
+      def readpartial: (*untyped args) -> String
 
       private
 
@@ -821,7 +821,7 @@ module HTTP
     def initialize: (Request req, Options options) -> void
     def failed_proxy_connect?: () -> bool
     def send_request: (Request req) -> void
-    def readpartial: (?Integer size) -> String?
+    def readpartial: (?Integer size, ?String? outbuf) -> String
     def read_headers!: () -> void
     def finish_response: () -> void
     def close: () -> void
@@ -836,7 +836,6 @@ module HTTP
 
     def init_state: (Options options) -> void
     def check_premature_eof: (bool eof) -> void
-    def body_framed?: () -> bool
     def connect_socket: (Request req, Options options) -> void
     def start_tls: (untyped req, untyped options) -> void
     def send_proxy_connect_request: (untyped req) -> void
@@ -858,6 +857,7 @@ module HTTP
       def handle_proxy_connect_response: () -> void
       def reset_timer: () -> void
       def set_keep_alive: () -> void
+      def body_framed?: () -> bool
       def read_more: (Integer size) -> untyped
     end
   end

test/http/connection_test.rb

@@ -280,13 +280,50 @@
 
       conn.read_headers!
       buffer = +""
-      while (s = conn.readpartial(3))
-        refute_predicate conn, :finished_request? if s != ""
-        buffer << s
+      begin
+        loop do
+          s = conn.readpartial(3)
+          refute_predicate conn, :finished_request? if s != ""
+          buffer << s
+        end
+      rescue EOFError
+        # Expected — end of response
       end
 
       assert_equal "1234567890", buffer
       assert_predicate conn, :finished_request?
     end
+
+    it "raises EOFError when no response is pending" do
+      assert_raises(EOFError) { connection.readpartial }
+    end
+
+    it "fills outbuf when provided" do
+      call_count = 0
+      responses = [
+        "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello",
+        :eof
+      ]
+      ob_socket = fake(
+        connect:     nil,
+        close:       nil,
+        readpartial: proc {
+          idx = [call_count, responses.length - 1].min
+          responses[idx].tap { call_count += 1 }
+        },
+        closed?:     proc { call_count >= responses.length }
+      )
+      ob_timeout_class = fake(new: ob_socket)
+      ob_opts = HTTP::Options.new(timeout_class: ob_timeout_class)
+      conn = HTTP::Connection.new(req, ob_opts)
+      conn.instance_variable_set(:@pending_response, true)
+
+      conn.read_headers!
+      outbuf = +""
+      result = conn.readpartial(16_384, outbuf)
+
+      assert_equal "hello", outbuf
+      assert_same outbuf, result
+    end
   end
 end

test/http/response/body_test.rb

@@ -7,7 +7,7 @@
   let(:body) { HTTP::Response::Body.new(connection, encoding: Encoding::UTF_8) }
 
   let(:connection) do
-    fake(sequence_id: 0, readpartial: proc { chunks.shift }, body_completed?: proc {
+    fake(sequence_id: 0, readpartial: proc { chunks.shift || raise(EOFError) }, body_completed?: proc {
       chunks.empty?
     })
   end
@@ -141,18 +141,25 @@
       it "streams decoded body" do
         assert_equal "Hi, HTTP ", body.readpartial
         assert_equal "here \u263A", body.readpartial
-        assert_nil body.readpartial
+        assert_raises(EOFError) { body.readpartial }
       end
     end
   end
 
-  context "when inflater receives nil chunk without prior data" do
-    it "closes the zstream and handles subsequent nil" do
-      conn = fake(sequence_id: 0, readpartial: proc {}, body_completed?: proc { true })
+  context "when inflater receives EOFError without prior data" do
+    it "closes the zstream and re-raises" do
+      conn = fake(readpartial: proc { raise EOFError })
       inflater = HTTP::Response::Inflater.new(conn)
-      inflater.readpartial
 
-      assert_nil inflater.readpartial
+      assert_raises(EOFError) { inflater.readpartial }
+    end
+
+    it "handles repeated EOFError after zstream is already closed" do
+      conn = fake(readpartial: proc { raise EOFError })
+      inflater = HTTP::Response::Inflater.new(conn)
+
+      assert_raises(EOFError) { inflater.readpartial }
+      assert_raises(EOFError) { inflater.readpartial }
     end
   end
 end

test/http/response_test.rb

@@ -357,7 +357,7 @@
     end
 
     let(:connection) do
-      fake(sequence_id: 0, readpartial: proc { chunks.shift }, body_completed?: proc {
+      fake(sequence_id: 0, readpartial: proc { chunks.shift || raise(EOFError) }, body_completed?: proc {
         chunks.empty?
       })
     end