Merge 8f0c5b8 into 341b23f

lib/reel.rb

@@ -11,9 +11,11 @@
 require 'reel/mixins'
 require 'reel/connection'
 require 'reel/logger'
+require 'reel/request_info'
 require 'reel/request'
-require 'reel/request_parser'
 require 'reel/response'
+require 'reel/request_parser'
+require 'reel/response_writer'
 require 'reel/server'
 require 'reel/ssl_server'
 require 'reel/websocket'

lib/reel/connection.rb

@@ -10,18 +10,20 @@ class StateError < RuntimeError; end # wrong state for a given operation
     TRANSFER_ENCODING  = 'Transfer-Encoding'.freeze
     KEEP_ALIVE         = 'Keep-Alive'.freeze
     CLOSE              = 'close'.freeze
-    CHUNKED            = 'chunked'.freeze
 
     attr_reader :socket, :parser
 
     # Attempt to read this much data
     BUFFER_SIZE = 16384
+    attr_reader :buffer_size
 
-    def initialize(socket)
+    def initialize(socket, buffer_size = nil)
       @attached  = true
       @socket    = socket
       @keepalive = true
-      @parser    = Request::Parser.new
+      @parser    = Request::Parser.new(socket, self)
+      @writer    = Response::Writer.new(socket, self)
+      @buffer_size = buffer_size.nil? ? BUFFER_SIZE : buffer_size
       reset_request
 
       @response_state = :header
@@ -40,21 +42,32 @@ def detach
     end
 
     # Reset the current request state
-    def reset_request(state = :header)
+    def reset_request(state = :ready)
       @request_state = state
-      @header_buffer = "" # Buffer headers in case of an upgrade request
+      @current_request = nil
       @parser.reset
     end
 
+    def readpartial(size = @buffer_size)
+      raise StateError, "can't read in the '#{@request_state}' request state" unless @request_state == :ready
+      @parser.readpartial(size)
+    end
+
+    def current_request
+      @current_request
+    end
+
     # Read a request object from the connection
     def request
       return if @request_state == :websocket
-      req = Request.read(self)
+      raise StateError, "current request not responded to" if current_request
+      req = @parser.current_request
 
       case req
       when Request
-        @request_state = :body
+        @request_state = :ready
         @keepalive = false if req[CONNECTION] == CLOSE || req.version == HTTP_VERSION_1_0
+        @current_request = req
       when WebSocket
         @request_state = @response_state = :websocket
         @socket = SocketUpgradedError
@@ -69,45 +82,10 @@ def request
       nil
     end
 
-    # Read a chunk from the request
-    def readpartial(size = BUFFER_SIZE)
-      raise StateError, "can't read in the `#{@request_state}' state" unless @request_state == :body
-
-      chunk = @parser.chunk
-      unless chunk || @parser.finished?
-        @parser << @socket.readpartial(size)
-        chunk = @parser.chunk
-      end
-
-      chunk
-    end
-
-    # read length bytes from request body
-    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? ? 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
-
     # Send a response back to the client
     # Response can be a symbol indicating the status code or a Reel::Response
     def respond(response, headers_or_body = {}, body = nil)
-      raise StateError "not in header state" if @response_state != :header
+      raise StateError, "not in header state" if @response_state != :header
 
       if headers_or_body.is_a? Hash
         headers = headers_or_body
@@ -129,18 +107,18 @@ def respond(response, headers_or_body = {}, body = nil)
       else raise TypeError, "invalid response: #{response.inspect}"
       end
 
-      response.render(@socket)
+      @writer.handle_response(response)
 
       # Enable streaming mode
-      if response.headers[TRANSFER_ENCODING] == CHUNKED and response.body.nil?
+      if response.chunked? and response.body.nil?
         @response_state = :chunked_body
       end
     rescue IOError, Errno::ECONNRESET, Errno::EPIPE
       # The client disconnected early
       @keepalive = false
     ensure
       if @keepalive
-        reset_request(:header)
+        reset_request(:ready)
       else
         @socket.close unless @socket.closed?
         reset_request(:closed)
@@ -150,21 +128,20 @@ def respond(response, headers_or_body = {}, body = nil)
     # Write body chunks directly to the connection
     def write(chunk)
       raise StateError, "not in chunked body mode" unless @response_state == :chunked_body
-      chunk_header = chunk.bytesize.to_s(16)
-      @socket << chunk_header + Response::CRLF
-      @socket << chunk + Response::CRLF
+      @writer.write(chunk)
     end
     alias_method :<<, :write
 
     # Finish the response and reset the response state to header
     def finish_response
       raise StateError, "not in body state" if @response_state != :chunked_body
-      @socket << "0#{Response::CRLF * 2}"
+      @writer.finish_response
       @response_state = :header
     end
 
     # Close the connection
     def close
+      raise StateError, "connection upgraded to Reel::WebSocket, call close on the websocket instance" if @response_state == :websocket
       @keepalive = false
       @socket.close unless @socket.closed?
     end

lib/reel/mixins.rb

@@ -24,23 +24,23 @@ def remote_host
   module RequestMixin
 
     def method
-      @http_parser.http_method
+      @request_info.http_method
     end
 
     def headers
-      @http_parser.headers
+      @request_info.headers
     end
 
     def [] header
       headers[header]
     end
 
     def version
-      @http_parser.http_version || HTTPVersionsMixin::DEFAULT_HTTP_VERSION
+      @request_info.http_version || HTTPVersionsMixin::DEFAULT_HTTP_VERSION
     end
 
     def url
-      @http_parser.url
+      @request_info.url
     end
 
     def uri

lib/reel/request.rb

@@ -5,52 +5,120 @@ class Request
     extend Forwardable
     include RequestMixin
 
-    UPGRADE   = 'Upgrade'.freeze
-    WEBSOCKET = 'websocket'.freeze
+    def self.build(request_info, connection)
+      request_info.method ||
+        raise(ArgumentError, "Unknown Request Method: %s" % request_info.http_method)
 
-    # Array#include? seems slow compared to Hash lookup
-    request_methods = Http::METHODS.map { |m| m.to_s.upcase }
-    REQUEST_METHODS = Hash[request_methods.zip(request_methods)].freeze
+      if request_info.websocket_request?
+        WebSocket.new(request_info, connection.socket)
+      else
+        Request.new(request_info, connection)
+      end
+    end
 
-    def self.read(connection)
-      parser = connection.parser
+    def_delegators :@connection, :<<, :write, :respond, :finish_response
 
-      begin
-        data = connection.socket.readpartial(Connection::BUFFER_SIZE)
-        parser << data
-      end until parser.headers
+    # request_info is a RequestInfo object including the headers and
+    # the url, method and http version.
+    #
+    # Access it through the RequestMixin methods.
+    def initialize(request_info, connection = nil)
+      @request_info = request_info
+      @connection = connection
+      @finished_read = false
+    end
+
+    # 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!
+      @finished_read = true
+    end
+
+    # Buffer body sent from connection, or send it directly to
+    # the @on_body callback if set (calling #body with a block)
+    def add_body(chunk)
+      if @on_body
+        @on_body.call(chunk)
+      else
+        @body ||= ""
+        @body << chunk
+      end
+    end
+
+    # Returns the body, if a block is given, the body is streamed
+    # to the block as the chunks become available, until the body
+    # has been read.
+    #
+    # If no block is given, the entire body will be read from the
+    # connection into the body buffer and then returned.
+    def body(&block)
+      raise "no connection given" unless @connection
+
+      if block_given?
+        # Callback from the http_parser will be calling add_body directly
+        @on_body = Proc.new(&block)
 
-      REQUEST_METHODS[parser.http_method] ||
-        raise(ArgumentError, "Unknown Request Method: %s" % parser.http_method)
+        # clear out body buffered so far
+        yield read_from_body(nil) if @body
 
-      upgrade = parser.headers[UPGRADE]
-      if upgrade && upgrade.downcase == WEBSOCKET
-        WebSocket.new(parser, connection.socket)
+        until finished_reading?
+          @connection.readpartial
+        end
+        @on_body = nil
       else
-        Request.new(parser, connection)
+        until finished_reading?
+          @connection.readpartial
+        end
+        @body
       end
     end
 
-    def_delegators :@connection, :respond, :finish_response, :close, :read
+    # Read a number of bytes, looping until they are available or until
+    # read_from_body returns nil, indicating there are no more bytes to read
+    #
+    # Note that bytes read from the body buffer will be cleared as they are
+    # 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
 
-    def initialize(http_parser, connection = nil)
-      @http_parser, @connection = http_parser, connection
+      chunk_size = length.nil? ? @connection.buffer_size : length
+      begin
+        while chunk_size > 0
+          chunk = read_from_body(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 body
-      @body ||= begin
-        raise "no connection given" unless @connection
-
-        body = "" unless block_given?
-        while (chunk = @connection.readpartial)
-          if block_given?
-            yield chunk
-          else
-            body << chunk
-          end
+    # @private
+    # Reads a number of bytes from the byte buffer, asking
+    # the connection to add to the buffer if there are not enough
+    # bytes available.
+    #
+    # Body buffer is cleared as bytes are read from it.
+    def read_from_body(length = nil)
+      if length.nil?
+        slice = @body
+        @body = nil
+      else
+        @body ||= ''
+        unless finished_reading? || @body.length >= length
+          @connection.readpartial(length - @body.length)
         end
-        body unless block_given?
+        slice = @body.slice!(0...length)
       end
+      slice && slice.length == 0 ? nil : slice
     end
+    private :read_from_body
+
   end
 end

lib/reel/request_info.rb

@@ -0,0 +1,27 @@
+module Reel
+  class RequestInfo
+    attr_reader :http_method, :url, :http_version, :headers
+
+    def initialize(http_method, url, http_version, headers)
+      @http_method  = http_method
+      @url          = url
+      @http_version = http_version
+      @headers      = headers
+    end
+
+    UPGRADE   = 'Upgrade'.freeze
+    WEBSOCKET = 'websocket'.freeze
+
+    # Array#include? seems slow compared to Hash lookup
+    request_methods = Http::METHODS.map { |m| m.to_s.upcase }
+    REQUEST_METHODS = Hash[request_methods.zip(request_methods)].freeze
+
+    def method
+      REQUEST_METHODS[http_method]
+    end
+
+    def websocket_request?
+      headers[UPGRADE] && headers[UPGRADE].downcase == WEBSOCKET
+    end
+  end
+end