Fix RequestProcessor connection reuse with unconsumed requests

spec/std/http/server/server_spec.cr

@@ -639,42 +639,211 @@ module HTTP
       ))
     end
 
-    it "skips body between requests" do
-      processor = HTTP::Server::RequestProcessor.new do |context|
-        context.response.content_type = "text/plain"
-        context.response.puts "Hello world\r"
+    describe "reads consecutive requests" do
+      it "when body is consumed" do
+        processor = HTTP::Server::RequestProcessor.new do |context|
+          context.response.content_type = "text/plain"
+          context.response << context.request.body.not_nil!.gets(chomp: true)
+          context.response << "\r\n"
+        end
+
+        input = IO::Memory.new(requestize(<<-REQUEST
+          POST / HTTP/1.1
+          Content-Length: 7
+
+          hello
+          POST / HTTP/1.1
+          Content-Length: 7
+
+          hello
+          REQUEST
+        ))
+        output = IO::Memory.new
+        processor.process(input, output)
+        output.rewind
+        output.gets_to_end.should eq(requestize(<<-RESPONSE
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Type: text/plain
+          Content-Length: 7
+
+          hello
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Type: text/plain
+          Content-Length: 7
+
+          hello
+
+          RESPONSE
+        ))
+      end
+
+      it "with empty body" do
+        processor = HTTP::Server::RequestProcessor.new do |context|
+          context.response.content_type = "text/plain"
+          context.response.puts "Hello world\r"
+        end
+
+        input = IO::Memory.new(requestize(<<-REQUEST
+          POST / HTTP/1.1
+
+          POST / HTTP/1.1
+          Content-Length: 7
+
+          hello
+          REQUEST
+        ))
+        output = IO::Memory.new
+        processor.process(input, output)
+        output.rewind
+        output.gets_to_end.should eq(requestize(<<-RESPONSE
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Type: text/plain
+          Content-Length: 13
+
+          Hello world
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Type: text/plain
+          Content-Length: 13
+
+          Hello world
+
+          RESPONSE
+        ))
+      end
+
+      it "fail if body is not consumed" do
+        processor = HTTP::Server::RequestProcessor.new do |context|
+          context.response.content_type = "text/plain"
+          context.response.puts "Hello world\r"
+        end
+
+        input = IO::Memory.new(requestize(<<-REQUEST
+          POST / HTTP/1.1
+
+          hello
+          POST / HTTP/1.1
+          Content-Length: 7
+
+          hello
+          REQUEST
+        ))
+        output = IO::Memory.new
+        processor.process(input, output)
+        output.rewind
+        output.gets_to_end.should eq(requestize(<<-RESPONSE
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Type: text/plain
+          Content-Length: 13
+
+          Hello world
+          HTTP/1.1 400 Bad Request
+          Content-Type: text/plain
+          Transfer-Encoding: chunked
+
+          10
+          400 Bad Request\\n
+          0
+
+
+          RESPONSE
+        ).gsub("\\n", "\n"))
       end
 
-      input = IO::Memory.new(requestize(<<-REQUEST
-        POST / HTTP/1.1
-        Content-Length: 7
+      it "closes connection when Connection: close" do
+        processor = HTTP::Server::RequestProcessor.new do |context|
+          context.response.headers["Connection"] = "close"
+        end
 
-        hello
-        POST / HTTP/1.1
-        Content-Length: 7
+        input = IO::Memory.new(requestize(<<-REQUEST
+          POST / HTTP/1.1
+          Content-Length: 7
 
-        hello
-        REQUEST
-      ))
-      output = IO::Memory.new
-      processor.process(input, output)
-      output.rewind
-      output.gets_to_end.should eq(requestize(<<-RESPONSE
-        HTTP/1.1 200 OK
-        Connection: keep-alive
-        Content-Type: text/plain
-        Content-Length: 13
+          hello
+          POST / HTTP/1.1
+          Content-Length: 7
 
-        Hello world
-        HTTP/1.1 200 OK
-        Connection: keep-alive
-        Content-Type: text/plain
-        Content-Length: 13
+          hello
+          REQUEST
+        ))
+        output = IO::Memory.new
+        processor.process(input, output)
+        output.rewind
+        output.gets_to_end.should eq(requestize(<<-RESPONSE
+          HTTP/1.1 200 OK
+          Connection: close
+          Content-Length: 0
 
-        Hello world
 
-        RESPONSE
-      ))
+          RESPONSE
+        ))
+      end
+
+      it "closes connection when request body is not entirely consumed" do
+        processor = HTTP::Server::RequestProcessor.new do |context|
+        end
+
+        input = IO::Memory.new(requestize(<<-REQUEST
+          POST / HTTP/1.1
+          Content-Length: 4
+
+          1
+          POST / HTTP/1.1
+          Content-Length: 7
+
+          hello
+          REQUEST
+        ))
+        output = IO::Memory.new
+        processor.process(input, output)
+        output.rewind
+        output.gets_to_end.should eq(requestize(<<-RESPONSE
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Length: 0
+
+
+          RESPONSE
+        ))
+      end
+
+      it "continues when request body is entirely consumed" do
+        processor = HTTP::Server::RequestProcessor.new do |context|
+          io = context.request.body.not_nil!
+          io.gets_to_end
+        end
+
+        input = IO::Memory.new(requestize(<<-REQUEST
+          POST / HTTP/1.1
+          Content-Length: 16387
+
+          #{"0" * 16_384}1
+          POST / HTTP/1.1
+          Content-Length: 7
+
+          hello
+          REQUEST
+        ))
+        output = IO::Memory.new
+        processor.process(input, output)
+        output.rewind
+        output.gets_to_end.should eq(requestize(<<-RESPONSE
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Length: 0
+
+          HTTP/1.1 200 OK
+          Connection: keep-alive
+          Content-Length: 0
+
+
+          RESPONSE
+        ))
+      end
     end
 
     it "handles Errno" do

src/http/content.cr

@@ -10,7 +10,6 @@ module HTTP
 
     def close
       @expects_continue = false
-      skip_to_end
       super
     end
 
@@ -217,5 +216,9 @@ module HTTP
     def write(slice : Bytes)
       raise IO::Error.new "Can't write to ChunkedContent"
     end
+
+    def closed?
+      @received_final_chunk || super
+    end
   end
 end

src/http/server.cr

@@ -113,6 +113,19 @@ require "./common"
 # Currently processing requests are not interrupted but also not waited for.
 # In order to give them some grace period for finishing, the calling context
 # can add a timeout like `sleep 10.seconds` after `#listen` returns.
+#
+# ### Reusing connections
+#
+# The request processor supports reusing a connection for subsequent
+# requests. This is used by default for HTTP/1.1 or when requested by
+# the `Connection: keep-alive` header. This is signalled by this header being
+# set on the `HTTP::Server::Response` when it's passed into the handler chain.
+#
+# If in the handler chain this header is overridden to `Connection: close`, then
+# the connection will not be reused after the request has been processed.
+#
+# Reusing the connection also requires that the request body (if present) is
+# entirely consumed in the handler chain. Otherwise the connection will be closed.
 class HTTP::Server
   @sockets = [] of Socket::Server
 

src/http/server/request_processor.cr

@@ -55,9 +55,22 @@ class HTTP::Server::RequestProcessor
 
         break unless request.keep_alive?
 
-        # Skip request body in case the handler
-        # didn't read it all, for the next request
-        request.body.try &.close
+        # Don't continue if the handler set `Connection` header to `close`
+        break unless HTTP.keep_alive?(response)
+
+        # The request body is either FixedLengthContent or ChunkedContent.
+        # In case it has not entirely been consumed by the handler, the connection is
+        # closed the connection even if keep alive was requested.
+        case body = request.body
+        when FixedLengthContent
+          if body.read_remaining > 0
+            # Close the connection if there are bytes remaining
+            break
+          end
+        when ChunkedContent
+          # Close the connection if the IO has still bytes to read.
+          break unless body.closed?
+        end
       end
     rescue ex : Errno
       # IO-related error, nothing to do