Merge pull request #409 from janko-m/enable-request-body-streaming

Enable request body streaming with an IO object

Author: ixti

http.gemspec

@@ -26,7 +26,7 @@ Gem::Specification.new do |gem|
   gem.required_ruby_version = ">= 2.0"
 
   gem.add_runtime_dependency "http_parser.rb", "~> 0.6.0"
-  gem.add_runtime_dependency "http-form_data", "~> 1.0.1"
+  gem.add_runtime_dependency "http-form_data", ">= 2.0.0-pre2", "< 3"
   gem.add_runtime_dependency "http-cookie",    "~> 1.0"
   gem.add_runtime_dependency "addressable",    "~> 2.3"
 

lib/http/client.rb

@@ -157,9 +157,8 @@ def make_request_body(opts, headers)
           opts.body
         when opts.form
           form = HTTP::FormData.create opts.form
-          headers[Headers::CONTENT_TYPE]   ||= form.content_type
-          headers[Headers::CONTENT_LENGTH] ||= form.content_length
-          form.to_s
+          headers[Headers::CONTENT_TYPE] ||= form.content_type
+          form
         when opts.json
           body = MimeType[:json].encode opts.json
           headers[Headers::CONTENT_TYPE] ||= "application/json; charset=#{body.encoding.name}"

lib/http/request.rb

@@ -74,7 +74,7 @@ class UnsupportedSchemeError < RequestError; end
     # @option opts [HTTP::URI, #to_s] :uri
     # @option opts [Hash] :headers
     # @option opts [Hash] :proxy
-    # @option opts [String] :body
+    # @option opts [String, Enumerable, IO, nil] :body
     def initialize(opts)
       @verb   = opts.fetch(:verb).to_s.downcase.to_sym
       @uri    = normalize_uri(opts.fetch(:uri))

lib/http/request/body.rb

@@ -0,0 +1,60 @@
+# frozen_string_literal: true
+
+module HTTP
+  class Request
+    class Body
+      # Maximum chunk size used for reading content of an IO
+      BUFFER_SIZE = Connection::BUFFER_SIZE
+
+      def initialize(body)
+        @body = body
+
+        validate_body_type!
+      end
+
+      # Returns size which should be used for the "Content-Length" header.
+      #
+      # @return [Integer]
+      def size
+        if @body.is_a?(String)
+          @body.bytesize
+        elsif @body.respond_to?(:read)
+          raise RequestError, "IO object must respond to #size" unless @body.respond_to?(:size)
+          @body.size
+        elsif @body.nil?
+          0
+        else
+          raise RequestError, "cannot determine size of body: #{@body.inspect}"
+        end
+      end
+
+      # Yields chunks of content to be streamed to the request body.
+      #
+      # @yieldparam [String]
+      def each
+        return enum_for(__method__) unless block_given?
+
+        if @body.is_a?(String)
+          yield @body
+        elsif @body.respond_to?(:read)
+          while (data = @body.read(BUFFER_SIZE))
+            yield data
+          end
+        elsif @body.is_a?(Enumerable)
+          @body.each { |chunk| yield chunk }
+        end
+      end
+
+      private
+
+      def validate_body_type!
+        return if @body.is_a?(String)
+        return if @body.respond_to?(:read)
+        return if @body.is_a?(Enumerable)
+        return if @body.nil?
+
+        raise RequestError, "body of wrong type: #{@body.class}"
+      end
+    end
+  end
+end

lib/http/request/writer.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 require "http/headers"
+require "http/request/body"
 
 module HTTP
   class Request
@@ -16,16 +17,11 @@ class Writer
       # End of a chunked transfer
       CHUNKED_END = "#{ZERO}#{CRLF}#{CRLF}".freeze
 
-      # Types valid to be used as body source
-      VALID_BODY_TYPES = [String, NilClass, Enumerable].freeze
-
       def initialize(socket, body, headers, headline)
-        @body           = body
+        @body           = Body.new(body)
         @socket         = socket
         @headers        = headers
         @request_header = [headline]
-
-        validate_body_type!
       end
 
       # Adds headers to the request header from the headers array
@@ -51,13 +47,9 @@ def connect_through_proxy
       # Adds the headers to the header array for the given request body we are working
       # with
       def add_body_type_headers
-        if @body.is_a?(String) && !@headers[Headers::CONTENT_LENGTH]
-          @request_header << "#{Headers::CONTENT_LENGTH}: #{@body.bytesize}"
-        elsif @body.nil? && !@headers[Headers::CONTENT_LENGTH]
-          @request_header << "#{Headers::CONTENT_LENGTH}: 0"
-        elsif @body.is_a?(Enumerable) && CHUNKED != @headers[Headers::TRANSFER_ENCODING]
-          raise(RequestError, "invalid transfer encoding")
-        end
+        return if @headers[Headers::CONTENT_LENGTH] || chunked?
+
+        @request_header << "#{Headers::CONTENT_LENGTH}: #{@body.size}"
       end
 
       # Joins the headers specified in the request into a correctly formatted
@@ -70,28 +62,42 @@ def join_headers
 
       def send_request
         headers = join_headers
+        chunks = @body.each
 
         # It's important to send the request in a single write call when
         # possible in order to play nicely with Nagle's algorithm. Making
         # two writes in a row triggers a pathological case where Nagle is
         # expecting a third write that never happens.
-        case @body
-        when NilClass
-          write(headers)
-        when String
-          write(headers << @body)
-        when Enumerable
+        begin
+          first_chunk = encode_chunk(chunks.next)
+          write(headers << first_chunk)
+        rescue StopIteration
           write(headers)
+        end
 
-          @body.each do |chunk|
-            write(chunk.bytesize.to_s(16) << CRLF << chunk << CRLF)
-          end
+        # Kernel#loop automatically rescues StopIteration
+        loop do
+          data = encode_chunk(chunks.next)
+          write(data)
+        end
 
-          write(CHUNKED_END)
-        else raise TypeError, "invalid body type: #{@body.class}"
+        write(CHUNKED_END) if chunked?
+      end
+
+      # Returns the chunk encoded for to the specified "Transfer-Encoding" header.
+      def encode_chunk(chunk)
+        if chunked?
+          chunk.bytesize.to_s(16) << CRLF << chunk << CRLF
+        else
+          chunk
         end
       end
 
+      # Returns true if the request should be sent in chunked encoding.
+      def chunked?
+        @headers[Headers::TRANSFER_ENCODING] == CHUNKED
+      end
+
       private
 
       def write(data)
@@ -101,11 +107,6 @@ def write(data)
           data = data.byteslice(length..-1)
         end
       end
-
-      def validate_body_type!
-        return if VALID_BODY_TYPES.any? { |type| @body.is_a? type }
-        raise RequestError, "body of wrong type: #{@body.class}"
-      end
     end
   end
 end

spec/lib/http/client_spec.rb

@@ -158,6 +158,32 @@ def simple_response(body, status = 200)
     end
   end
 
+  describe "passing multipart form data" do
+    it "creates url encoded form data object" do
+      client = HTTP::Client.new
+      allow(client).to receive(:perform)
+
+      expect(HTTP::Request).to receive(:new) do |opts|
+        expect(opts[:body]).to be_a(HTTP::FormData::Urlencoded)
+        expect(opts[:body].to_s).to eq "foo=bar"
+      end
+
+      client.get("http://example.com/", :form => {:foo => "bar"})
+    end
+
+    it "creates multipart form data object" do
+      client = HTTP::Client.new
+      allow(client).to receive(:perform)
+
+      expect(HTTP::Request).to receive(:new) do |opts|
+        expect(opts[:body]).to be_a(HTTP::FormData::Multipart)
+        expect(opts[:body].to_s).to include("content")
+      end
+
+      client.get("http://example.com/", :form => {:foo => HTTP::FormData::Part.new("content")})
+    end
+  end
+
   describe "passing json" do
     it "encodes given object" do
       client = HTTP::Client.new