Skip to content

Commit

Permalink
Close inactive connections and requests
Browse files Browse the repository at this point in the history
This new middleware introduces a timeout of closing inactive
connections between requests after a configured amount of seconds.

This builds on top of reactphp#405 and partially on reactphp#422
  • Loading branch information
WyriHaximus committed Sep 15, 2022
1 parent 00f3590 commit 25f7136
Show file tree
Hide file tree
Showing 8 changed files with 253 additions and 20 deletions.
17 changes: 17 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ multiple concurrent HTTP requests without blocking.
* [ServerRequest](#serverrequest)
* [ResponseException](#responseexception)
* [React\Http\Middleware](#reacthttpmiddleware)
* [InactiveConnectionTimeoutMiddleware](#inactiveconnectiontimeoutmiddleware)
* [StreamingRequestMiddleware](#streamingrequestmiddleware)
* [LimitConcurrentRequestsMiddleware](#limitconcurrentrequestsmiddleware)
* [RequestBodyBufferMiddleware](#requestbodybuffermiddleware)
Expand Down Expand Up @@ -2630,6 +2631,22 @@ access its underlying response object.

### React\Http\Middleware

#### InactiveConnectionTimeoutMiddleware

The `React\Http\Middleware\InactiveConnectionTimeoutMiddleware` is purely a configuration middleware to configure the
`HttpServer` to close any inactive connections between requests to close the connection and not leave them needlessly open.

The following example configures the `HttpServer` to close any inactive connections after one and a half second:

```php
$http = new React\Http\HttpServer(
new React\Http\Middleware\InactiveConnectionTimeoutMiddleware(1.5),
$handler
);
```
> Internally, this class is used as a "value object" to override the default timeout of one minute.
As such it doesn't have any behavior internally, that is all in the internal "StreamingServer".

#### StreamingRequestMiddleware

The `React\Http\Middleware\StreamingRequestMiddleware` can be used to
Expand Down
13 changes: 10 additions & 3 deletions src/HttpServer.php
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,12 @@
use Evenement\EventEmitter;
use React\EventLoop\Loop;
use React\EventLoop\LoopInterface;
use React\Http\Io\Clock;
use React\Http\Io\IniUtil;
use React\Http\Io\MiddlewareRunner;
use React\Http\Io\RequestHeaderParser;
use React\Http\Io\StreamingServer;
use React\Http\Middleware\InactiveConnectionTimeoutMiddleware;
use React\Http\Middleware\LimitConcurrentRequestsMiddleware;
use React\Http\Middleware\StreamingRequestMiddleware;
use React\Http\Middleware\RequestBodyBufferMiddleware;
Expand Down Expand Up @@ -219,10 +222,13 @@ public function __construct($requestHandlerOrLoop)
}

$streaming = false;
$idleConnectTimeout = InactiveConnectionTimeoutMiddleware::DEFAULT_TIMEOUT;
foreach ((array) $requestHandlers as $handler) {
if ($handler instanceof StreamingRequestMiddleware) {
$streaming = true;
break;
}
if ($handler instanceof InactiveConnectionTimeoutMiddleware) {
$idleConnectTimeout = $handler->getTimeout();
}
}

Expand Down Expand Up @@ -252,10 +258,11 @@ public function __construct($requestHandlerOrLoop)
* doing anything with the request.
*/
$middleware = \array_filter($middleware, function ($handler) {
return !($handler instanceof StreamingRequestMiddleware);
return !($handler instanceof StreamingRequestMiddleware) && !($handler instanceof InactiveConnectionTimeoutMiddleware);
});

$this->streamingServer = new StreamingServer($loop, new MiddlewareRunner($middleware));
$clock = new Clock($loop);
$this->streamingServer = new StreamingServer(new MiddlewareRunner($middleware), new RequestHeaderParser($loop, $clock, $idleConnectTimeout), $clock);

$that = $this;
$this->streamingServer->on('error', function ($error) use ($that) {
Expand Down
38 changes: 35 additions & 3 deletions src/Io/RequestHeaderParser.php
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

use Evenement\EventEmitter;
use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\LoopInterface;
use React\Http\Message\Response;
use React\Http\Message\ServerRequest;
use React\Socket\ConnectionInterface;
Expand All @@ -24,23 +25,53 @@ class RequestHeaderParser extends EventEmitter
{
private $maxSize = 8192;

/**
* @var LoopInterface
*/
private $loop;

/** @var Clock */
private $clock;

/**
* @var float
*/
private $idleConnectionTimeout;

/** @var array<string|int,array<string,string>> */
private $connectionParams = array();

public function __construct(Clock $clock)
/**
* @param LoopInterface $loop
* @param float $idleConnectionTimeout
*/
public function __construct(LoopInterface $loop, Clock $clock, $idleConnectionTimeout)
{
$this->loop = $loop;
$this->clock = $clock;
$this->idleConnectionTimeout = $idleConnectionTimeout;
}

public function handle(ConnectionInterface $conn)
{
$loop = $this->loop;
$idleConnectionTimeout = $this->idleConnectionTimeout;
$that = $this;
$idleConnectionTimeoutHandler = function () use ($that, $conn) {
$that->emit('error', array(
new \RuntimeException('Request timed out', Response::STATUS_REQUEST_TIMEOUT),
$conn
));
$conn->close();
};
$timer = $loop->addTimer($idleConnectionTimeout, $idleConnectionTimeoutHandler);
$conn->on('close', function () use ($loop, &$timer) {
$loop->cancelTimer($timer);
});
$buffer = '';
$maxSize = $this->maxSize;
$that = $this;
$conn->on('data', $fn = function ($data) use (&$buffer, &$fn, $conn, $maxSize, $that) {
$conn->on('data', $fn = function ($data) use (&$buffer, &$fn, $conn, $maxSize, $that, $loop, &$timer, $idleConnectionTimeout, $idleConnectionTimeoutHandler) {
$loop->cancelTimer($timer);
// append chunk of data to buffer and look for end of request headers
$buffer .= $data;
$endOfHeader = \strpos($buffer, "\r\n\r\n");
Expand All @@ -59,6 +90,7 @@ public function handle(ConnectionInterface $conn)

// ignore incomplete requests
if ($endOfHeader === false) {
$timer = $loop->addTimer($idleConnectionTimeout, $idleConnectionTimeoutHandler);
return;
}

Expand Down
15 changes: 8 additions & 7 deletions src/Io/StreamingServer.php
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
use Evenement\EventEmitter;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\LoopInterface;
use React\Http\Message\Response;
use React\Http\Message\ServerRequest;
use React\Promise;
Expand All @@ -28,7 +27,7 @@
* object in return:
*
* ```php
* $server = new StreamingServer($loop, function (ServerRequestInterface $request) {
* $server = new StreamingServer(function (ServerRequestInterface $request) {
* return new Response(
* Response::STATUS_OK,
* array(
Expand All @@ -53,7 +52,7 @@
* in order to start a plaintext HTTP server like this:
*
* ```php
* $server = new StreamingServer($loop, $handler);
* $server = new StreamingServer($handler);
*
* $socket = new React\Socket\SocketServer('0.0.0.0:8080', array(), $loop);
* $server->listen($socket);
Expand Down Expand Up @@ -86,6 +85,8 @@ final class StreamingServer extends EventEmitter

/** @var Clock */
private $clock;
private $loop;
private $idleConnectionTimeout;

/**
* Creates an HTTP server that invokes the given callback for each incoming HTTP request
Expand All @@ -95,19 +96,19 @@ final class StreamingServer extends EventEmitter
* connections in order to then parse incoming data as HTTP.
* See also [listen()](#listen) for more details.
*
* @param LoopInterface $loop
* @param callable $requestHandler
* @param float $idleConnectTimeout
* @see self::listen()
*/
public function __construct(LoopInterface $loop, $requestHandler)
public function __construct($requestHandler, RequestHeaderParser $parser, Clock $clock)
{
if (!\is_callable($requestHandler)) {
throw new \InvalidArgumentException('Invalid request handler given');
}

$this->callback = $requestHandler;
$this->clock = new Clock($loop);
$this->parser = new RequestHeaderParser($this->clock);
$this->clock = $clock;
$this->parser = $parser;

$that = $this;
$this->parser->on('headers', function (ServerRequestInterface $request, ConnectionInterface $conn) use ($that) {
Expand Down
62 changes: 62 additions & 0 deletions src/Middleware/InactiveConnectionTimeoutMiddleware.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
<?php

namespace React\Http\Middleware;

use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\ServerRequestInterface;
use React\Http\Io\HttpBodyStream;
use React\Http\Io\PauseBufferStream;
use React\Promise;
use React\Promise\PromiseInterface;
use React\Promise\Deferred;
use React\Stream\ReadableStreamInterface;

/**
* Closes any inactive connection after the specified amount of seconds since last activity.
*
* This allows you to set an alternative timeout to the default one minute (60 seconds). For example
* thirteen and a half seconds:
*
* ```php
* $http = new React\Http\HttpServer(
* new React\Http\Middleware\InactiveConnectionTimeoutMiddleware(13.5),
* $handler
* );
*
* > Internally, this class is used as a "value object" to override the default timeout of one minute.
* As such it doesn't have any behavior internally, that is all in the internal "StreamingServer".
*/
final class InactiveConnectionTimeoutMiddleware
{
/**
* @internal
*/
const DEFAULT_TIMEOUT = 60;

/**
* @var float
*/
private $timeout;

/**
* @param float $timeout
*/
public function __construct($timeout = self::DEFAULT_TIMEOUT)
{
$this->timeout = $timeout;
}

public function __invoke(ServerRequestInterface $request, $next)
{
return $next($request);
}

/**
* @return float
* @internal
*/
public function getTimeout()
{
return $this->timeout;
}
}
24 changes: 23 additions & 1 deletion tests/HttpServerTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
use React\EventLoop\Loop;
use React\Http\HttpServer;
use React\Http\Io\IniUtil;
use React\Http\Io\StreamingServer;
use React\Http\Middleware\InactiveConnectionTimeoutMiddleware;
use React\Http\Middleware\StreamingRequestMiddleware;
use React\Promise;
use React\Promise\Deferred;
Expand Down Expand Up @@ -60,10 +62,18 @@ public function testConstructWithoutLoopAssignsLoopAutomatically()
$ref->setAccessible(true);
$clock = $ref->getValue($streamingServer);

$ref = new \ReflectionProperty($streamingServer, 'parser');
$ref->setAccessible(true);
$parser = $ref->getValue($streamingServer);

$ref = new \ReflectionProperty($clock, 'loop');
$ref->setAccessible(true);
$loop = $ref->getValue($clock);

$ref = new \ReflectionProperty($parser, 'loop');
$ref->setAccessible(true);
$loop = $ref->getValue($parser);

$this->assertInstanceOf('React\EventLoop\LoopInterface', $loop);
}

Expand Down Expand Up @@ -257,6 +267,18 @@ function (ServerRequestInterface $request) use (&$streaming) {
$this->assertEquals(true, $streaming);
}

public function testIdleConnectionWillBeClosedAfterConfiguredTimeout()
{
$this->connection->expects($this->once())->method('close');

$http = new HttpServer(Loop::get(), new InactiveConnectionTimeoutMiddleware(0.1), $this->expectCallableNever());

$http->listen($this->socket);
$this->socket->emit('connection', array($this->connection));

Loop::run();
}

public function testForwardErrors()
{
$exception = new \Exception();
Expand Down Expand Up @@ -439,7 +461,7 @@ public function testConstructServerWithMemoryLimitDoesLimitConcurrency()

public function testConstructFiltersOutConfigurationMiddlewareBefore()
{
$http = new HttpServer(new StreamingRequestMiddleware(), function () { });
$http = new HttpServer(new InactiveConnectionTimeoutMiddleware(0), new StreamingRequestMiddleware(), function () { });

$ref = new \ReflectionProperty($http, 'streamingServer');
$ref->setAccessible(true);
Expand Down
Loading

0 comments on commit 25f7136

Please sign in to comment.