Skip to content

Commit c919cb7

Browse files
WyriHaximusWyriHaximus
authored
Merge pull request #405 from clue-labs/keepalive
Support persistent connections (Connection: keep-alive)
2 parents 5944a08 + 84fbe78 commit c919cb7

6 files changed

+249
-24
lines changed

‎README.md

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1698,11 +1698,12 @@ so you don't have to. For instance, if the client sends the request using the
16981698
HTTP/1.1 protocol version, the response message will also use the same protocol
16991699
version, no matter what version is returned from the request handler function.
17001700

1701-
Note that persistent connections (`Connection: keep-alive`) are currently
1702-
not supported.
1703-
As such, HTTP/1.1 response messages will automatically include a
1704-
`Connection: close` header, irrespective of what header values are
1705-
passed explicitly.
1701+
The server supports persistent connections. An appropriate `Connection: keep-alive`
1702+
or `Connection: close` response header will be added automatically, respecting the
1703+
matching request header value and HTTP default header values. The server is
1704+
responsible for handling the `Connection` response header, so you SHOULD NOT pass
1705+
this response header yourself, unless you explicitly want to override the user's
1706+
choice with a `Connection: close` response header.
17061707

17071708
### Middleware
17081709

‎examples/99-server-benchmark-download.php

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,18 @@
11
<?php
22

3+
// A simple HTTP web server that can be used to benchmark requests per second and download speed
4+
//
35
// $ php examples/99-server-benchmark-download.php 8080
6+
//
7+
// This example runs the web server on a single CPU core in order to measure the
8+
// per core performance.
9+
//
410
// $ curl http://localhost:8080/10g.bin > /dev/null
511
// $ wget http://localhost:8080/10g.bin -O /dev/null
6-
// $ ab -n10 -c10 http://localhost:8080/1g.bin
7-
// $ docker run -it --rm --net=host jordi/ab -n100000 -c10 http://localhost:8080/
8-
// $ docker run -it --rm --net=host jordi/ab -n10 -c10 http://localhost:8080/1g.bin
12+
// $ ab -n10 -c10 -k http://localhost:8080/1g.bin
13+
// $ docker run -it --rm --net=host jordi/ab -n100000 -c10 -k http://localhost:8080/
14+
// $ docker run -it --rm --net=host jordi/ab -n10 -c10 -k http://localhost:8080/1g.bin
15+
// $ docker run -it --rm --net=host skandyla/wrk -t8 -c10 -d20 http://localhost:8080/
916

1017
use Evenement\EventEmitter;
1118
use Psr\Http\Message\ServerRequestInterface;

‎src/Io/RequestHeaderParser.php

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -106,10 +106,6 @@ public function handle(ConnectionInterface $conn)
106106
$stream->close();
107107
}
108108
});
109-
110-
$conn->on('close', function () use (&$buffer, &$fn) {
111-
$fn = $buffer = null;
112-
});
113109
}
114110

115111
/**

‎src/Io/StreamingServer.php

Lines changed: 35 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -210,7 +210,8 @@ public function writeError(ConnectionInterface $conn, $code, ServerRequestInterf
210210
$response = new Response(
211211
$code,
212212
array(
213-
'Content-Type' => 'text/plain'
213+
'Content-Type' => 'text/plain',
214+
'Connection' => 'close' // we do not want to keep the connection open after an error
214215
),
215216
'Error ' . $code
216217
);
@@ -273,17 +274,28 @@ public function handleResponse(ConnectionInterface $connection, ServerRequestInt
273274
$chunked = true;
274275
} else {
275276
// remove any Transfer-Encoding headers unless automatically enabled above
277+
// we do not want to keep connection alive, so pretend we received "Connection: close" request header
276278
$response = $response->withoutHeader('Transfer-Encoding');
279+
$request = $request->withHeader('Connection', 'close');
277280
}
278281

279282
// assign "Connection" header automatically
283+
$persist = false;
280284
if ($code === 101) {
281285
// 101 (Switching Protocols) response uses Connection: upgrade header
286+
// This implies that this stream now uses another protocol and we
287+
// may not persist this connection for additional requests.
282288
$response = $response->withHeader('Connection', 'upgrade');
283-
} elseif ($version === '1.1') {
284-
// HTTP/1.1 assumes persistent connection support by default
285-
// we do not support persistent connections, so let the client know
289+
} elseif (\strtolower($request->getHeaderLine('Connection')) === 'close' || \strtolower($response->getHeaderLine('Connection')) === 'close') {
290+
// obey explicit "Connection: close" request header or response header if present
286291
$response = $response->withHeader('Connection', 'close');
292+
} elseif ($version === '1.1') {
293+
// HTTP/1.1 assumes persistent connection support by default, so we don't need to inform client
294+
$persist = true;
295+
} elseif (strtolower($request->getHeaderLine('Connection')) === 'keep-alive') {
296+
// obey explicit "Connection: keep-alive" request header and inform client
297+
$persist = true;
298+
$response = $response->withHeader('Connection', 'keep-alive');
287299
} else {
288300
// remove any Connection headers unless automatically enabled above
289301
$response = $response->withoutHeader('Connection');
@@ -328,9 +340,15 @@ public function handleResponse(ConnectionInterface $connection, ServerRequestInt
328340
$body = "0\r\n\r\n";
329341
}
330342

331-
// end connection after writing response headers and body
343+
// write response headers and body
332344
$connection->write($headers . "\r\n" . $body);
333-
$connection->end();
345+
346+
// either wait for next request over persistent connection or end connection
347+
if ($persist) {
348+
$this->parser->handle($connection);
349+
} else {
350+
$connection->end();
351+
}
334352
return;
335353
}
336354

@@ -345,6 +363,16 @@ public function handleResponse(ConnectionInterface $connection, ServerRequestInt
345363
// in particular this may only fire on a later read/write attempt.
346364
$connection->on('close', array($body, 'close'));
347365

348-
$body->pipe($connection);
366+
// write streaming body and then wait for next request over persistent connection
367+
if ($persist) {
368+
$body->pipe($connection, array('end' => false));
369+
$parser = $this->parser;
370+
$body->on('end', function () use ($connection, $parser, $body) {
371+
$connection->removeListener('close', array($body, 'close'));
372+
$parser->handle($connection);
373+
});
374+
} else {
375+
$body->pipe($connection);
376+
}
349377
}
350378
}

‎tests/FunctionalServerTest.php

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -662,7 +662,7 @@ public function testConnectWithThroughStreamReturnsDataAsGiven()
662662
$server->listen($socket);
663663

664664
$result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) {
665-
$conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\n\r\n");
665+
$conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\nConnection: close\r\n\r\n");
666666

667667
$conn->once('data', function () use ($conn) {
668668
$conn->write('hello');
@@ -703,7 +703,7 @@ public function testConnectWithThroughStreamReturnedFromPromiseReturnsDataAsGive
703703
$server->listen($socket);
704704

705705
$result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) {
706-
$conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\n\r\n");
706+
$conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\nConnection: close\r\n\r\n");
707707

708708
$conn->once('data', function () use ($conn) {
709709
$conn->write('hello');
@@ -737,7 +737,7 @@ public function testConnectWithClosedThroughStreamReturnsNoData()
737737
$server->listen($socket);
738738

739739
$result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) {
740-
$conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\n\r\n");
740+
$conn->write("CONNECT example.com:80 HTTP/1.1\r\nHost: example.com:80\r\nConnection: close\r\n\r\n");
741741

742742
$conn->once('data', function () use ($conn) {
743743
$conn->write('hello');

‎tests/Io/StreamingServerTest.php

Lines changed: 195 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
use React\EventLoop\Factory;
77
use React\Http\Io\StreamingServer;
88
use React\Http\Message\Response;
9+
use React\Http\Message\ServerRequest;
910
use React\Promise\Promise;
1011
use React\Stream\ThroughStream;
1112
use React\Tests\Http\SocketServerStub;
@@ -957,7 +958,7 @@ function ($data) use (&$buffer) {
957958
$data = "GET / HTTP/1.1\r\n\r\n";
958959
$this->connection->emit('data', array($data));
959960

960-
$this->assertEquals("HTTP/1.1 200 OK\r\nUpgrade: demo\r\nContent-Length: 3\r\nConnection: close\r\n\r\nfoo", $buffer);
961+
$this->assertEquals("HTTP/1.1 200 OK\r\nUpgrade: demo\r\nContent-Length: 3\r\n\r\nfoo", $buffer);
961962
}
962963

963964
public function testResponseUpgradeWishInRequestCanBeIgnoredByReturningNormalResponse()
@@ -992,7 +993,7 @@ function ($data) use (&$buffer) {
992993
$data = "GET / HTTP/1.1\r\nUpgrade: demo\r\n\r\n";
993994
$this->connection->emit('data', array($data));
994995

995-
$this->assertEquals("HTTP/1.1 200 OK\r\nContent-Length: 3\r\nConnection: close\r\n\r\nfoo", $buffer);
996+
$this->assertEquals("HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nfoo", $buffer);
996997
}
997998

998999
public function testResponseUpgradeSwitchingProtocolIncludesConnectionUpgradeHeaderWithoutContentLength()
@@ -2813,6 +2814,198 @@ public function testRequestCookieWithCommaValueWillBeAddedToServerRequest() {
28132814
$this->assertEquals(array('test' => 'abc,def', 'hello' => 'world'), $requestValidation->getCookieParams());
28142815
}
28152816

2817+
public function testNewConnectionWillInvokeParserOnce()
2818+
{
2819+
$server = new StreamingServer(Factory::create(), $this->expectCallableNever());
2820+
2821+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2822+
$parser->expects($this->once())->method('handle');
2823+
2824+
$ref = new \ReflectionProperty($server, 'parser');
2825+
$ref->setAccessible(true);
2826+
$ref->setValue($server, $parser);
2827+
2828+
$server->listen($this->socket);
2829+
$this->socket->emit('connection', array($this->connection));
2830+
}
2831+
2832+
public function testNewConnectionWillInvokeParserOnceAndInvokeRequestHandlerWhenParserIsDoneForHttp10()
2833+
{
2834+
$request = new ServerRequest('GET', 'http://localhost/', array(), '', '1.0');
2835+
2836+
$server = new StreamingServer(Factory::create(), $this->expectCallableOnceWith($request));
2837+
2838+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2839+
$parser->expects($this->once())->method('handle');
2840+
2841+
$ref = new \ReflectionProperty($server, 'parser');
2842+
$ref->setAccessible(true);
2843+
$ref->setValue($server, $parser);
2844+
2845+
$server->listen($this->socket);
2846+
$this->socket->emit('connection', array($this->connection));
2847+
2848+
$this->connection->expects($this->once())->method('write');
2849+
$this->connection->expects($this->once())->method('end');
2850+
2851+
// pretend parser just finished parsing
2852+
$server->handleRequest($this->connection, $request);
2853+
}
2854+
2855+
public function testNewConnectionWillInvokeParserOnceAndInvokeRequestHandlerWhenParserIsDoneForHttp11ConnectionClose()
2856+
{
2857+
$request = new ServerRequest('GET', 'http://localhost/', array('Connection' => 'close'));
2858+
2859+
$server = new StreamingServer(Factory::create(), $this->expectCallableOnceWith($request));
2860+
2861+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2862+
$parser->expects($this->once())->method('handle');
2863+
2864+
$ref = new \ReflectionProperty($server, 'parser');
2865+
$ref->setAccessible(true);
2866+
$ref->setValue($server, $parser);
2867+
2868+
$server->listen($this->socket);
2869+
$this->socket->emit('connection', array($this->connection));
2870+
2871+
$this->connection->expects($this->once())->method('write');
2872+
$this->connection->expects($this->once())->method('end');
2873+
2874+
// pretend parser just finished parsing
2875+
$server->handleRequest($this->connection, $request);
2876+
}
2877+
2878+
public function testNewConnectionWillInvokeParserOnceAndInvokeRequestHandlerWhenParserIsDoneAndRequestHandlerReturnsConnectionClose()
2879+
{
2880+
$request = new ServerRequest('GET', 'http://localhost/');
2881+
2882+
$server = new StreamingServer(Factory::create(), function () {
2883+
return new Response(200, array('Connection' => 'close'));
2884+
});
2885+
2886+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2887+
$parser->expects($this->once())->method('handle');
2888+
2889+
$ref = new \ReflectionProperty($server, 'parser');
2890+
$ref->setAccessible(true);
2891+
$ref->setValue($server, $parser);
2892+
2893+
$server->listen($this->socket);
2894+
$this->socket->emit('connection', array($this->connection));
2895+
2896+
$this->connection->expects($this->once())->method('write');
2897+
$this->connection->expects($this->once())->method('end');
2898+
2899+
// pretend parser just finished parsing
2900+
$server->handleRequest($this->connection, $request);
2901+
}
2902+
2903+
public function testNewConnectionWillInvokeParserTwiceAfterInvokingRequestHandlerWhenConnectionCanBeKeptAliveForHttp11Default()
2904+
{
2905+
$request = new ServerRequest('GET', 'http://localhost/');
2906+
2907+
$server = new StreamingServer(Factory::create(), function () {
2908+
return new Response();
2909+
});
2910+
2911+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2912+
$parser->expects($this->exactly(2))->method('handle');
2913+
2914+
$ref = new \ReflectionProperty($server, 'parser');
2915+
$ref->setAccessible(true);
2916+
$ref->setValue($server, $parser);
2917+
2918+
$server->listen($this->socket);
2919+
$this->socket->emit('connection', array($this->connection));
2920+
2921+
$this->connection->expects($this->once())->method('write');
2922+
$this->connection->expects($this->never())->method('end');
2923+
2924+
// pretend parser just finished parsing
2925+
$server->handleRequest($this->connection, $request);
2926+
}
2927+
2928+
public function testNewConnectionWillInvokeParserTwiceAfterInvokingRequestHandlerWhenConnectionCanBeKeptAliveForHttp10ConnectionKeepAlive()
2929+
{
2930+
$request = new ServerRequest('GET', 'http://localhost/', array('Connection' => 'keep-alive'), '', '1.0');
2931+
2932+
$server = new StreamingServer(Factory::create(), function () {
2933+
return new Response();
2934+
});
2935+
2936+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2937+
$parser->expects($this->exactly(2))->method('handle');
2938+
2939+
$ref = new \ReflectionProperty($server, 'parser');
2940+
$ref->setAccessible(true);
2941+
$ref->setValue($server, $parser);
2942+
2943+
$server->listen($this->socket);
2944+
$this->socket->emit('connection', array($this->connection));
2945+
2946+
$this->connection->expects($this->once())->method('write');
2947+
$this->connection->expects($this->never())->method('end');
2948+
2949+
// pretend parser just finished parsing
2950+
$server->handleRequest($this->connection, $request);
2951+
}
2952+
2953+
public function testNewConnectionWillInvokeParserOnceAfterInvokingRequestHandlerWhenStreamingResponseBodyKeepsStreaming()
2954+
{
2955+
$request = new ServerRequest('GET', 'http://localhost/');
2956+
2957+
$body = new ThroughStream();
2958+
$server = new StreamingServer(Factory::create(), function () use ($body) {
2959+
return new Response(200, array(), $body);
2960+
});
2961+
2962+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2963+
$parser->expects($this->once())->method('handle');
2964+
2965+
$ref = new \ReflectionProperty($server, 'parser');
2966+
$ref->setAccessible(true);
2967+
$ref->setValue($server, $parser);
2968+
2969+
$server->listen($this->socket);
2970+
$this->socket->emit('connection', array($this->connection));
2971+
2972+
$this->connection->expects($this->once())->method('write');
2973+
$this->connection->expects($this->never())->method('end');
2974+
2975+
// pretend parser just finished parsing
2976+
$server->handleRequest($this->connection, $request);
2977+
}
2978+
2979+
public function testNewConnectionWillInvokeParserTwiceAfterInvokingRequestHandlerWhenStreamingResponseBodyEnds()
2980+
{
2981+
$request = new ServerRequest('GET', 'http://localhost/');
2982+
2983+
$body = new ThroughStream();
2984+
$server = new StreamingServer(Factory::create(), function () use ($body) {
2985+
return new Response(200, array(), $body);
2986+
});
2987+
2988+
$parser = $this->getMockBuilder('React\Http\Io\RequestHeaderParser')->getMock();
2989+
$parser->expects($this->exactly(2))->method('handle');
2990+
2991+
$ref = new \ReflectionProperty($server, 'parser');
2992+
$ref->setAccessible(true);
2993+
$ref->setValue($server, $parser);
2994+
2995+
$server->listen($this->socket);
2996+
$this->socket->emit('connection', array($this->connection));
2997+
2998+
$this->connection->expects($this->exactly(2))->method('write');
2999+
$this->connection->expects($this->never())->method('end');
3000+
3001+
// pretend parser just finished parsing
3002+
$server->handleRequest($this->connection, $request);
3003+
3004+
$this->assertCount(2, $this->connection->listeners('close'));
3005+
$body->end();
3006+
$this->assertCount(1, $this->connection->listeners('close'));
3007+
}
3008+
28163009
private function createGetRequest()
28173010
{
28183011
$data = "GET / HTTP/1.1\r\n";

0 commit comments

Comments
 (0)