Repository: reactphp/http (php)

Issue:
Support persistent connections (aka HTTP/1.1 Keep-Alive)

Pull request description:
Support persistent connections (Connection: keep-alive)
This changeset finally adds support for persistent connections (Connection: keep-alive) 🎉

This is a pure feature addition that should show a noticeable performance improvement when using persistent connections (which is the default pretty much everywhere). Together with #404, this improves benchmarking performance from ~11000 requests per second to ~22000 requests per second on a single CPU core.

```bash
$ docker run -it --rm --net=host jordi/ab -n100000 -c80 -k http://localhost:8080/
…
Requests per second:    22327.67 [#/sec] (mean)
…
```

This has limited support for pipelining (which is rarely used in practice), I'll look into this in a follow-up PR because this requires a considerable refactoring first. If you do not wish to use persistent connections, you can always use a middleware like this:

```php
$server = new Server(
    $loop,
    function (ServerRequestInterface $request, callable $next) {
        // disable keep-alive (not usually recommended)
        return $next($request)->withHeader('Connection', 'close');
    },
    function (ServerRequestInterface $request) {
        return new Response(
            200,
            []
            'Hello world!'
        );
    }
);
```

Resolves #39

Patch:
diff --git a/README.md b/README.md
index 09a4ecd..4625a81 100644
--- a/README.md
+++ b/README.md
@@ -1698,11 +1698,12 @@ so you don't have to. For instance, if the client sends the request using the
 HTTP/1.1 protocol version, the response message will also use the same protocol
 version, no matter what version is returned from the request handler function.
 
-Note that persistent connections (`Connection: keep-alive`) are currently
-not supported.
-As such, HTTP/1.1 response messages will automatically include a
-`Connection: close` header, irrespective of what header values are
-passed explicitly.
+The server supports persistent connections. An appropriate `Connection: keep-alive`
+or `Connection: close` response header will be added automatically, respecting the
+matching request header value and HTTP default header values. The server is
+responsible for handling the `Connection` response header, so you SHOULD NOT pass
+this response header yourself, unless you explicitly want to override the user's
+choice with a `Connection: close` response header.
 
 ### Middleware
 
diff --git a/examples/99-server-benchmark-download.php b/examples/99-server-benchmark-download.php
index 5fdd55c..1a49df7 100644
--- a/examples/99-server-benchmark-download.php
+++ b/examples/99-server-benchmark-download.php
@@ -1,11 +1,18 @@
 <?php
 
+// A simple HTTP web server that can be used to benchmark requests per second and download speed
+//
 // $ php examples/99-server-benchmark-download.php 8080
+//
+// This example runs the web server on a single CPU core in order to measure the
+// per core performance.
+//
 // $ curl http://localhost:8080/10g.bin > /dev/null
 // $ wget http://localhost:8080/10g.bin -O /dev/null
-// $ ab -n10 -c10 http://localhost:8080/1g.bin
-// $ docker run -it --rm --net=host jordi/ab -n100000 -c10 http://localhost:8080/
-// $ docker run -it --rm --net=host jordi/ab -n10 -c10 http://localhost:8080/1g.bin
+// $ ab -n10 -c10 -k http://localhost:8080/1g.bin
+// $ docker run -it --rm --net=host jordi/ab -n100000 -c10 -k http://localhost:8080/
+// $ docker run -it --rm --net=host jordi/ab -n10 -c10 -k http://localhost:8080/1g.bin
+// $ docker run -it --rm --net=host skandyla/wrk -t8 -c10 -d20 http://localhost:8080/
 
 use Evenement\EventEmitter;
 use Psr\Http\Message\ServerRequestInterface;
diff --git a/src/Io/RequestHeaderParser.php b/src/Io/RequestHeaderParser.php
index 53f7ff0..5125c77 100644
--- a/src/Io/RequestHeaderParser.php
+++ b/src/Io/RequestHeaderParser.php
@@ -106,10 +106,6 @@ class RequestHeaderParser extends EventEmitter
                 $stream->close();
             }
         });
-
-        $conn->on('close', function () use (&$buffer, &$fn) {
-            $fn = $buffer = null;
-        });
     }
 
     /**
diff --git a/src/Io/StreamingServer.php b/src/Io/StreamingServer.php
index 0674d96..e20ddf4 100644
--- a/src/Io/StreamingServer.php
+++ b/src/Io/StreamingServer.php
@@ -210,7 +210,8 @@ final class StreamingServer extends EventEmitter
         $response = new Response(
             $code,
             array(
-                'Content-Type' => 'text/plain'
+                'Content-Type' => 'text/plain',
+                'Connection' => 'close' // we do not want to keep the connection open after an error
             ),
             'Error ' . $code
         );
@@ -273,17 +274,28 @@ final class StreamingServer extends EventEmitter
             $chunked = true;
         } else {
             // remove any Transfer-Encoding headers unless automatically enabled above
+            // we do not want to keep connection alive, so pretend we received "Connection: close" request header
             $response = $response->withoutHeader('Transfer-Encoding');
+            $request = $request->withHeader('Connection', 'close');
         }
 
         // assign "Connection" header automatically
+        $persist = false;
         if ($code === 101) {
             // 101 (Switching Protocols) response uses Connection: upgrade header
+            // This implies that this stream now uses another protocol and we
+            // may not persist this connection for additional requests.
             $response = $response->withHeader('Connection', 'upgrade');
-        } elseif ($version === '1.1') {
-            // HTTP/1.1 assumes persistent connection support by default
-            // we do not support persistent connections, so let the client know
+        } elseif (\strtolower($request->getHeaderLine('Connection')) === 'close' || \strtolower($response->getHeaderLine('Connection')) === 'close') {
+            // obey explicit "Connection: close" request header or response header if present
             $response = $response->withHeader('Connection', 'close');
+        } elseif ($version === '1.1') {
+            // HTTP/1.1 assumes persistent connection support by default, so we don't need to inform client
+            $persist = true;
+        } elseif (strtolower($request->getHeaderLine('Connection')) === 'keep-alive') {
+            // obey explicit "Connection: keep-alive" request header and inform client
+            $persist = true;
+            $response = $response->withHeader('Connection', 'keep-alive');
         } else {
             // remove any Connection headers unless automatically enabled above
             $response = $response->withoutHeader('Connection');
@@ -328,9 +340,15 @@ final class StreamingServer extends EventEmitter
                 $body = "0\r\n\r\n";
             }
 
-            // end connection after writing response headers and body
+            // write response headers and body
             $connection->write($headers . "\r\n" . $body);
-            $connection->end();
+
+            // either wait for next request over persistent connection or end connection
+            if ($persist) {
+                $this->parser->handle($connection);
+            } else {
+                $connection->end();
+            }
             return;
         }
 
@@ -345,6 +363,16 @@ final class StreamingServer extends EventEmitter
         // in particular this may only fire on a later read/write attempt.
         $connection->on('close', array($body, 'close'));
 
-        $body->pipe($connection);
+        // write streaming body and then wait for next request over persistent connection
+        if ($persist) {
+            $body->pipe($connection, array('end' => false));
+            $parser = $this->parser;
+            $body->on('end', function () use ($connection, $parser, $body) {
+                $connection->removeListener('close', array($body, 'close'));
+                $parser->handle($connection);
+            });
+        } else {
+            $body->pipe($connection);
+        }
     }
 }


Test patch:
diff --git a/tests/FunctionalServerTest.php b/tests/FunctionalServerTest.php
index bd127ab..41cf31d 100644
--- a/tests/FunctionalServerTest.php
+++ b/tests/FunctionalServerTest.php
@@ -662,7 +662,7 @@ class FunctionalServerTest extends TestCase
         $server->listen($socket);
 
         $result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) {
-            $conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\n\r\n");
+            $conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\nConnection: close\r\n\r\n");
 
             $conn->once('data', function () use ($conn) {
                 $conn->write('hello');
@@ -703,7 +703,7 @@ class FunctionalServerTest extends TestCase
         $server->listen($socket);
 
         $result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) {
-            $conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\n\r\n");
+            $conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\nConnection: close\r\n\r\n");
 
             $conn->once('data', function () use ($conn) {
                 $conn->write('hello');
@@ -737,7 +737,7 @@ class FunctionalServerTest extends TestCase
         $server->listen($socket);
 
         $result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) {
-            $conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\n\r\n");
+            $conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\nConnection: close\r\n\r\n");
 
             $conn->once('data', function () use ($conn) {
                 $conn->write('hello');
diff --git a/tests/Io/StreamingServerTest.php b/tests/Io/StreamingServerTest.php
index d2401a0..0dde7a0 100644
--- a/tests/Io/StreamingServerTest.php
+++ b/tests/Io/StreamingServerTest.php
@@ -6,6 +6,7 @@ use Psr\Http\Message\ServerRequestInterface;
 use React\EventLoop\Factory;
 use React\Http\Io\StreamingServer;
 use React\Http\Message\Response;
+use React\Http\Message\ServerRequest;
 use React\Promise\Promise;
 use React\Stream\ThroughStream;
 use React\Tests\Http\SocketServerStub;
@@ -957,7 +958,7 @@ class StreamingServerTest extends TestCase
         $data = "GET / HTTP/1.1\r\n\r\n";
         $this->connection->emit('data', array($data));
 
-        $this->assertEquals("HTTP/1.1 200 OK\r\nUpgrade: demo\r\nContent-Length: 3\r\nConnection: close\r\n\r\nfoo", $buffer);
+        $this->assertEquals("HTTP/1.1 200 OK\r\nUpgrade: demo\r\nContent-Length: 3\r\n\r\nfoo", $buffer);
     }
 
     public function testResponseUpgradeWishInRequestCanBeIgnoredByReturningNormalResponse()
@@ -992,7 +993,7 @@ class StreamingServerTest extends TestCase
         $data = "GET / HTTP/1.1\r\nUpgrade: demo\r\n\r\n";
         $this->connection->emit('data', array($data));
 
-        $this->assertEquals("HTTP/1.1 200 OK\r\nContent-Length: 3\r\nConnection: close\r\n\r\nfoo", $buffer);
+        $this->assertEquals("HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nfoo", $buffer);
     }
 
     public function testResponseUpgradeSwitchingProtocolIncludesConnectionUpgradeHeaderWithoutContentLength()
@@ -2813,6 +2814,198 @@ class StreamingServerTest extends TestCase
         $this->assertEquals(array('test' => 'abc,def', 'hello' => 'world'), $requestValidation->getCookieParams());
     }
 
+    public function testNewConnectionWillInvokeParserOnce()
+    {
+        $server = new StreamingServer(Factory::create(), $this->expectCallableNever());
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->once())->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+    }
+
+    public function testNewConnectionWillInvokeParserOnceAndInvokeRequestHandlerWhenParserIsDoneForHttp10()
+    {
+        $request = new ServerRequest('GET', 'http://localhost/', array(), '', '1.0');
+
+        $server = new StreamingServer(Factory::create(), $this->expectCallableOnceWith($request));
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->once())->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+
+        $this->connection->expects($this->once())->method('write');
+        $this->connection->expects($this->once())->method('end');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($this->connection, $request);
+    }
+
+    public function testNewConnectionWillInvokeParserOnceAndInvokeRequestHandlerWhenParserIsDoneForHttp11ConnectionClose()
+    {
+        $request = new ServerRequest('GET', 'http://localhost/', array('Connection' => 'close'));
+
+        $server = new StreamingServer(Factory::create(), $this->expectCallableOnceWith($request));
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->once())->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+
+        $this->connection->expects($this->once())->method('write');
+        $this->connection->expects($this->once())->method('end');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($this->connection, $request);
+    }
+
+    public function testNewConnectionWillInvokeParserOnceAndInvokeRequestHandlerWhenParserIsDoneAndRequestHandlerReturnsConnectionClose()
+    {
+        $request = new ServerRequest('GET', 'http://localhost/');
+
+        $server = new StreamingServer(Factory::create(), function () {
+            return new Response(200, array('Connection' => 'close'));
+        });
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->once())->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+
+        $this->connection->expects($this->once())->method('write');
+        $this->connection->expects($this->once())->method('end');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($this->connection, $request);
+    }
+
+    public function testNewConnectionWillInvokeParserTwiceAfterInvokingRequestHandlerWhenConnectionCanBeKeptAliveForHttp11Default()
+    {
+        $request = new ServerRequest('GET', 'http://localhost/');
+
+        $server = new StreamingServer(Factory::create(), function () {
+            return new Response();
+        });
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->exactly(2))->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+
+        $this->connection->expects($this->once())->method('write');
+        $this->connection->expects($this->never())->method('end');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($this->connection, $request);
+    }
+
+    public function testNewConnectionWillInvokeParserTwiceAfterInvokingRequestHandlerWhenConnectionCanBeKeptAliveForHttp10ConnectionKeepAlive()
+    {
+        $request = new ServerRequest('GET', 'http://localhost/', array('Connection' => 'keep-alive'), '', '1.0');
+
+        $server = new StreamingServer(Factory::create(), function () {
+            return new Response();
+        });
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->exactly(2))->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+
+        $this->connection->expects($this->once())->method('write');
+        $this->connection->expects($this->never())->method('end');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($this->connection, $request);
+    }
+
+    public function testNewConnectionWillInvokeParserOnceAfterInvokingRequestHandlerWhenStreamingResponseBodyKeepsStreaming()
+    {
+        $request = new ServerRequest('GET', 'http://localhost/');
+
+        $body = new ThroughStream();
+        $server = new StreamingServer(Factory::create(), function () use ($body) {
+            return new Response(200, array(), $body);
+        });
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->once())->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+
+        $this->connection->expects($this->once())->method('write');
+        $this->connection->expects($this->never())->method('end');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($this->connection, $request);
+    }
+
+    public function testNewConnectionWillInvokeParserTwiceAfterInvokingRequestHandlerWhenStreamingResponseBodyEnds()
+    {
+        $request = new ServerRequest('GET', 'http://localhost/');
+
+        $body = new ThroughStream();
+        $server = new StreamingServer(Factory::create(), function () use ($body) {
+            return new Response(200, array(), $body);
+        });
+
+        $parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
+        $parser->expects($this->exactly(2))->method('handle');
+
+        $ref = new \ReflectionProperty($server, 'parser');
+        $ref->setAccessible(true);
+        $ref->setValue($server, $parser);
+
+        $server->listen($this->socket);
+        $this->socket->emit('connection', array($this->connection));
+
+        $this->connection->expects($this->exactly(2))->method('write');
+        $this->connection->expects($this->never())->method('end');
+
+        // pretend parser just finished parsing
+        $server->handleRequest($this->connection, $request);
+
+        $this->assertCount(2, $this->connection->listeners('close'));
+        $body->end();
+        $this->assertCount(1, $this->connection->listeners('close'));
+    }
+
     private function createGetRequest()
     {
         $data = "GET / HTTP/1.1\r\n";