Extract HTTP::Connection ala Reel (WIP: DO NOT MERGE)

lib/http.rb

@@ -2,6 +2,7 @@
 
 require 'http/chainable'
 require 'http/client'
+require 'http/connection'
 require 'http/options'
 require 'http/request'
 require 'http/request/writer'

lib/http/client.rb

@@ -14,7 +14,6 @@ class Client
 
     def initialize(default_options = {})
       @default_options = HTTP::Options.new(default_options)
-      @parser = HTTP::Response::Parser.new
       @socket = nil
     end
 
@@ -40,9 +39,7 @@ def perform(req, options)
       if options.follow
         res = Redirector.new(options.follow).perform req, res do |request|
           # TODO: keep-alive
-          @parser.reset
           finish_response
-
           perform_without_following_redirects request, options
         end
       end
@@ -51,39 +48,6 @@ def perform(req, options)
       res
     end
 
-    # Read a chunk of the body
-    def readpartial(size = BUFFER_SIZE) # rubocop:disable CyclomaticComplexity
-      if @parser.finished? || (@body_remaining && @body_remaining.zero?)
-        chunk = @parser.chunk
-
-        if !chunk && @body_remaining && !@body_remaining.zero?
-          fail StateError, "expected #{@body_remaining} more bytes of body"
-        end
-
-        @body_remaining -= chunk.bytesize if chunk
-        return chunk
-      end
-
-      fail StateError, 'not connected' unless @socket
-
-      chunk = @parser.chunk
-      unless chunk
-        @parser << @socket.readpartial(BUFFER_SIZE)
-        chunk = @parser.chunk
-
-        # TODO: consult @body_remaining here and raise if appropriate
-        return unless chunk
-      end
-
-      if @body_remaining
-        @body_remaining -= chunk.bytesize
-        @body_remaining = nil if @body_remaining < 1
-      end
-
-      finish_response if @parser.finished?
-      chunk
-    end
-
   private
 
     # Perform a single (no follow) HTTP request
@@ -93,17 +57,18 @@ def perform_without_following_redirects(req, options)
       # TODO: keep-alive support
       @socket = options[:socket_class].open(req.socket_host, req.socket_port)
       @socket = start_tls(@socket, options) if uri.is_a?(URI::HTTPS)
+      connection = HTTP::Connection.new(@socket)
+      parser     = HTTP::Parser.new(connection)
 
       req.stream @socket
 
       begin
-        @parser << @socket.readpartial(BUFFER_SIZE) until @parser.headers
+        parser << @socket.readpartial(BUFFER_SIZE) until parser.headers
       rescue IOError, Errno::ECONNRESET, Errno::EPIPE => ex
         raise IOError, "problem making HTTP request: #{ex}"
       end
 
-      body = Response::Body.new(self)
-      Response.new(@parser.status_code, @parser.http_version, @parser.headers, body, uri)
+      HTTP::Response.new(connection, parser.status_code, parser.http_version, parser.headers, uri)
     end
 
     # Initialize TLS connection

lib/http/connection.rb

@@ -0,0 +1,107 @@
+require 'http/response/parser'
+
+module HTTP
+  # A connection to the HTTP server
+  class Connection
+    CONNECTION         = 'Connection'.freeze
+    TRANSFER_ENCODING  = 'Transfer-Encoding'.freeze
+    KEEP_ALIVE         = 'Keep-Alive'.freeze
+    CLOSE              = 'close'.freeze
+
+    attr_reader   :socket, :parser, :current_response
+    attr_accessor :request_state, :response_state
+
+    # Attempt to read this much data
+    BUFFER_SIZE = 16384
+    attr_reader :buffer_size
+
+    def initialize(socket, buffer_size = nil)
+      @socket      = socket
+      @keepalive   = true
+      @buffer_size = buffer_size || BUFFER_SIZE
+      @parser      = Response::Parser.new(self)
+
+      @request_state  = :headers
+      @response_state = :headers
+      reset_response
+    end
+
+    # Is the connection still active?
+    def alive?; @keepalive; end
+
+    # Send a request to the server
+    # Response can be a symbol indicating the status code or a HTTP::Response
+    def respond(response, headers_or_body = {}, body = nil)
+      raise StateError, "not in header state" if @response_state != :headers
+
+      if headers_or_body.is_a? Hash
+        headers = headers_or_body
+      else
+        headers = {}
+        body = headers_or_body
+      end
+
+      if @keepalive
+        headers[CONNECTION] = KEEP_ALIVE
+      else
+        headers[CONNECTION] = CLOSE
+      end
+
+      case response
+      when Symbol
+        response = Response.new(response, headers, body)
+      when Response
+      else raise TypeError, "invalid response: #{response.inspect}"
+      end
+
+      if current_request
+        current_request.handle_response(response)
+      else
+        raise RequestError
+      end
+
+      # Enable streaming mode
+      if response.chunked? and response.body.nil?
+        @response_state = :chunked_body
+      elsif @keepalive
+        reset_request
+      else
+        @current_request = nil
+        @parser.reset
+        @request_state = :closed
+      end
+    rescue IOError, Errno::ECONNRESET, Errno::EPIPE, RequestError
+      # The client disconnected early, or there is no request
+      @keepalive = false
+      @request_state = :closed
+    end
+
+    def readpartial(size = @buffer_size)
+      unless @request_state == :headers || @request_state == :body
+        raise StateError, "can't read in the '#{@request_fsm.state}' request state"
+      end
+
+      @parser.readpartial(size)
+    end
+
+    # Close the connection
+    def close
+      @keepalive = false
+      @socket.close unless @socket.closed?
+    end
+
+    # Reset the current response state
+    def reset_response
+      @response_state  = :headers
+      @current_request = nil
+      @parser.reset
+    end
+    private :reset_response
+
+    # Set response state for the connection.
+    def response_state=(state)
+      reset_response if state == :headers
+      @response_state = state
+    end
+  end
+end

lib/http/response.rb

@@ -74,8 +74,14 @@ class Response
     alias_method :code,        :status
     alias_method :status_code, :status
 
-    def initialize(status, version, headers, body, uri = nil) # rubocop:disable ParameterLists
-      @status, @version, @body, @uri = status, version, body, uri
+    def initialize(connection, status, version, headers, uri = nil) # rubocop:disable ParameterLists
+      @connection    = connection
+      @status        = status
+      @version       = version
+      @uri           = uri
+      @finished_read = false
+      @buffer        = ""
+      @body          = Response::Body.new(self)
 
       @headers = {}
       headers.each { |field, value| self[field] = value }
@@ -136,26 +142,66 @@ def charset
       @mime_type ||= content_type.charset
     end
 
-    # Inspect a response
-    def inspect
-      "#<#{self.class}/#{@version} #{status} #{reason} @headers=#{@headers.inspect}>"
+    # Returns true if request fully finished reading
+    def finished_reading?; @finished_read; end
+
+    # When HTTP Parser marks the message parsing as complete, this will be set.
+    def finish_reading!
+      raise StateError, "already finished" if @finished_read
+      @finished_read = true
     end
 
-    class BodyDelegator < ::Delegator
-      attr_reader :response
+    # Fill the request buffer with data as it becomes available
+    def fill_buffer(chunk)
+      @buffer << chunk
+    end
 
-      def initialize(response, body = response.body)
-        super(body)
-        @response, @body = response, body
+    # Read a number of bytes, looping until they are available or until
+    # readpartial returns nil, indicating there are no more bytes to read
+    def read(length = nil, buffer = nil)
+      raise ArgumentError, "negative length #{length} given" if length && length < 0
+
+      return '' if length == 0
+      res = buffer.nil? ? '' : buffer.clear
+
+      chunk_size = length.nil? ? @connection.buffer_size : length
+      begin
+        while chunk_size > 0
+          chunk = readpartial(chunk_size)
+          break unless chunk
+          res << chunk
+          chunk_size = length - res.length unless length.nil?
+        end
+      rescue EOFError
       end
+      return length && res.length == 0 ? nil : res
+    end
 
-      def __getobj__
-        @body
+    # Read a string up to the given number of bytes, blocking until some
+    # data is available but returning immediately if some data is available
+    def readpartial(length = nil)
+      if length.nil? && @buffer.length > 0
+        slice = @buffer
+        @buffer = ""
+      else
+        unless finished_reading? || (length && length <= @buffer.length)
+          @connection.readpartial(length ? length - @buffer.length : @connection.buffer_size)
+        end
+
+        if length
+          slice = @buffer.slice!(0, length)
+        else
+          slice = @buffer
+          @buffer = ""
+        end
       end
 
-      def __setobj__(obj)
-        @body = obj
-      end
+      slice && slice.length == 0 ? nil : slice
+    end
+
+    # Inspect a response
+    def inspect
+      "#<#{self.class}/#{@version} #{status} #{reason} @headers=#{@headers.inspect}>"
     end
   end
 end

lib/http/response/body.rb

@@ -8,22 +8,22 @@ class Body
       include Enumerable
       def_delegator :to_s, :empty?
 
-      def initialize(client)
-        @client    = client
+      def initialize(response)
+        @response  = response
         @streaming = nil
         @contents  = nil
       end
 
       # Read exactly the given amount of data
       def read(length)
         stream!
-        @client.read(length)
+        @response.read(length)
       end
 
       # Read up to length bytes, but return any data that's available
       def readpartial(length = nil)
         stream!
-        @client.readpartial(length)
+        @response.readpartial(length)
       end
 
       # Iterate over the body, allowing it to be enumerable
@@ -34,14 +34,14 @@ def each
       end
 
       # Eagerly consume the entire body as a string
-      def to_s
+      def to_str
         return @contents if @contents
         fail StateError, 'body is being streamed' unless @streaming.nil?
 
         begin
           @streaming = false
           @contents = ''
-          while (chunk = @client.readpartial)
+          while (chunk = @response.readpartial)
             @contents << chunk
           end
         rescue
@@ -51,18 +51,19 @@ def to_s
 
         @contents
       end
-      alias_method :to_str, :to_s
+      alias_method :to_s, :to_str
+
+      # Easier to interpret string inspect
+      def inspect
+        "#<#{self.class}:#{object_id.to_s(16)} @streaming=#{!!@streaming}>"
+      end
 
       # Assert that the body is actively being streamed
       def stream!
         fail StateError, 'body has already been consumed' if @streaming == false
         @streaming = true
       end
-
-      # Easier to interpret string inspect
-      def inspect
-        "#<#{self.class}:#{object_id.to_s(16)} @streaming=#{!!@streaming}>"
-      end
+      private :stream!
     end
   end
 end