Ensure LimitHandlersMiddleware pauses streaming requests

Author: clue

README.md

@@ -13,6 +13,7 @@ Event-driven, streaming plaintext HTTP and secure HTTPS server for [ReactPHP](ht
   * [Request](#request)
   * [Response](#response)
   * [Middleware](#middleware)
+    * [LimitHandlersMiddleware](#limithandlersmiddleware)
     * [RequestBodyBufferMiddleware](#requestbodybuffermiddleware)
     * [RequestBodyParserMiddleware](#requestbodyparsermiddleware)
     * [Third-Party Middleware](#third-party-middleware)
@@ -681,6 +682,59 @@ $server = new StreamingServer(new MiddlewareRunner([
 ]));
 ```
 
+#### LimitHandlersMiddleware
+
+The `LimitHandlersMiddleware` can be used to
+limit how many next handlers can be executed concurrently.
+
+If this middleware is invoked, it will check if the number of pending
+handlers is below the allowed limit and then simply invoke the next handler
+and it will return whatever the next handler returns (or throws).
+
+If the number of pending handlers exceeds the allowed limit, the request will
+be queued (and its streaming body will be paused) and it will return a pending
+promise.
+Once a pending handler returns (or throws), it will pick the oldest request
+from this queue and invokes the next handler (and its streaming body will be
+resumed).
+
+The following example shows how this middleware can be used to ensure no more
+than 10 handlers will be invoked at once:
+
+```php
+$server = new StreamingServer(new MiddlewareRunner([
+    new LimitHandlersMiddleware(10),
+    $handler
+]));
+```
+
+Similarly, this middleware is often used in combination with the
+[`RequestBodyBufferMiddleware`](#requestbodybuffermiddleware) (see below)
+to limit the total number of requests that can be buffered at once:
+
+```php
+$server = new StreamingServer(new MiddlewareRunner([
+    new LimitHandlersMiddleware(100), // 100 concurrent buffering handlers
+    new RequestBodyBufferMiddleware(2 * 1024 * 1024), // 2 MiB per request
+    new RequestBodyParserMiddleware(),
+    $handler
+]));
+```
+
+More sophisticated examples include limiting the total number of requests
+that can be buffered at once and then ensure the actual request handler only
+processes one request after another without any concurrency:
+
+```php
+$server = new StreamingServer(new MiddlewareRunner([
+    new LimitHandlersMiddleware(100), // 100 concurrent buffering handlers
+    new RequestBodyBufferMiddleware(2 * 1024 * 1024), // 2 MiB per request
+    new RequestBodyParserMiddleware(),
+    new LimitHandlersMiddleware(1), // only execute 1 handler (no concurrency)
+    $handler
+]));
+```
+
 #### RequestBodyBufferMiddleware
 
 One of the built-in middleware is the `RequestBodyBufferMiddleware` which
@@ -714,10 +768,18 @@ Similarly, this will immediately invoke the next middleware handler for requests
 that have an empty request body (such as a simple `GET` request) and requests
 that are already buffered (such as due to another middleware).
 
+Note that the given buffer size limit is applied to each request individually.
+This means that if you allow a 2 MiB limit and then receive 1000 concurrent
+requests, up to 2000 MiB may be allocated for these buffers alone.
+As such, it's highly recommended to use this along with the
+[`LimitHandlersMiddleware`](#limithandlersmiddleware) (see above) to limit
+the total number of concurrent requests.
+
 Usage:
 
 ```php
 $middlewares = new MiddlewareRunner([
+    new LimitHandlersMiddleware(100), // 100 concurrent buffering handlers
     new RequestBodyBufferMiddleware(16 * 1024 * 1024), // 16 MiB
     function (ServerRequestInterface $request, callable $next) {
         // The body from $request->getBody() is now fully available without the need to stream it 
@@ -776,6 +838,7 @@ $handler = function (ServerRequestInterface $request) {
 };
 
 $server = new StreamingServer(new MiddlewareRunner([
+    new LimitHandlersMiddleware(100), // 100 concurrent buffering handlers
     new RequestBodyBufferMiddleware(16 * 1024 * 1024), // 16 MiB
     new RequestBodyParserMiddleware(),
     $handler

examples/12-upload.php

@@ -11,6 +11,7 @@
 use Psr\Http\Message\UploadedFileInterface;
 use React\EventLoop\Factory;
 use React\Http\MiddlewareRunner;
+use React\Http\Middleware\LimitHandlersMiddleware;
 use React\Http\Middleware\RequestBodyBufferMiddleware;
 use React\Http\Middleware\RequestBodyParserMiddleware;
 use React\Http\Response;
@@ -121,6 +122,7 @@
 
 // buffer and parse HTTP request body before running our request handler
 $server = new StreamingServer(new MiddlewareRunner(array(
+    new LimitHandlersMiddleware(100), // 100 concurrent buffering handlers, queue otherwise
     new RequestBodyBufferMiddleware(8 * 1024 * 1024), // 8 MiB max, ignore body otherwise
     new RequestBodyParserMiddleware(100 * 1024, 1), // 1 file with 100 KiB max, reject upload otherwise
     $handler

src/Middleware/LimitHandlersMiddleware.php

@@ -3,10 +3,65 @@
 namespace React\Http\Middleware;
 
 use Psr\Http\Message\ServerRequestInterface;
+use React\Http\Io\HttpBodyStream;
+use React\Http\Io\PauseBufferStream;
 use React\Promise\Deferred;
 use React\Stream\ReadableStreamInterface;
 use SplQueue;
 
+/**
+ * Limits how many next handlers can be executed concurrently.
+ *
+ * If this middleware is invoked, it will check if the number of pending
+ * handlers is below the allowed limit and then simply invoke the next handler
+ * and it will return whatever the next handler returns (or throws).
+ *
+ * If the number of pending handlers exceeds the allowed limit, the request will
+ * be queued (and its streaming body will be paused) and it will return a pending
+ * promise.
+ * Once a pending handler returns (or throws), it will pick the oldest request
+ * from this queue and invokes the next handler (and its streaming body will be
+ * resumed).
+ *
+ * The following example shows how this middleware can be used to ensure no more
+ * than 10 handlers will be invoked at once:
+ *
+ * ```php
+ * $server = new StreamingServer(new MiddlewareRunner([
+ *     new LimitHandlersMiddleware(10),
+ *     $handler
+ * ]));
+ * ```
+ *
+ * Similarly, this middleware is often used in combination with the
+ * [`RequestBodyBufferMiddleware`](#requestbodybuffermiddleware) (see below)
+ * to limit the total number of requests that can be buffered at once:
+ *
+ * ```php
+ * $server = new StreamingServer(new MiddlewareRunner([
+ *     new LimitHandlersMiddleware(100), // 100 concurrent buffering handlers
+ *     new RequestBodyBufferMiddleware(2 * 1024 * 1024), // 2 MiB per request
+ *     new RequestBodyParserMiddleware(),
+ *    $handler
+ * ]));
+ * ```
+ *
+ * More sophisticated examples include limiting the total number of requests
+ * that can be buffered at once and then ensure the actual request handler only
+ * processes one request after another without any concurrency:
+ *
+ * ```php
+ * $server = new StreamingServer(new MiddlewareRunner([
+ *     new LimitHandlersMiddleware(100), // 100 concurrent buffering handlers
+ *     new RequestBodyBufferMiddleware(2 * 1024 * 1024), // 2 MiB per request
+ *     new RequestBodyParserMiddleware(),
+ *     new LimitHandlersMiddleware(1), // only execute 1 handler (no concurrency)
+ *     $handler
+ * ]));
+ * ```
+ *
+ * @see RequestBodyBufferMiddleware
+ */
 final class LimitHandlersMiddleware
 {
     private $limit;
@@ -29,6 +84,14 @@ public function __invoke(ServerRequestInterface $request, $next)
     {
         $body = $request->getBody();
         if ($body instanceof ReadableStreamInterface) {
+            // replace with buffering body to ensure any readable events will be buffered
+            $body = new HttpBodyStream(
+                new PauseBufferStream($body),
+                $body->getSize()
+            );
+
+            // pause actual body to stop emitting data until the handler is called
+            $request = $request->withBody($body);
             $body->pause();
         }
         $deferred = new Deferred();
@@ -40,11 +103,16 @@ public function __invoke(ServerRequestInterface $request, $next)
         $pending = &$this->pending;
         return $deferred->promise()->then(function () use ($request, $next, &$pending) {
             $pending++;
+
+            $ret = $next($request);
+
+            // resume readable stream and replay buffered events
             $body = $request->getBody();
             if ($body instanceof ReadableStreamInterface) {
                 $body->resume();
             }
-            return $next($request);
+
+            return $ret;
         })->then(function ($response) use ($that, &$pending) {
             $pending--;
             $that->processQueue();

tests/FunctionalServerTest.php

@@ -2,6 +2,9 @@
 
 namespace React\Tests\Http;
 
+use Psr\Http\Message\ServerRequestInterface;
+use React\Http\Middleware\LimitHandlersMiddleware;
+use React\Http\Middleware\RequestBodyBufferMiddleware;
 use React\Http\MiddlewareRunner;
 use React\Socket\Server as Socket;
 use React\EventLoop\Factory;
@@ -12,7 +15,7 @@
 use Clue\React\Block;
 use React\Http\Response;
 use React\Socket\SecureServer;
-use React\Promise\Promise;
+use React\Promise;
 use React\Promise\Stream;
 use React\Stream\ThroughStream;
 
@@ -653,7 +656,7 @@ public function testConnectWithThroughStreamReturnedFromPromiseReturnsDataAsGive
                 $stream->end();
             });
 
-            return new Promise(function ($resolve) use ($loop, $stream) {
+            return new Promise\Promise(function ($resolve) use ($loop, $stream) {
                 $loop->addTimer(0.001, function () use ($resolve, $stream) {
                     $resolve(new Response(200, array(), $stream));
                 });
@@ -715,6 +718,53 @@ public function testConnectWithClosedThroughStreamReturnsNoData()
 
         $socket->close();
     }
+
+    public function testLimitHandlersMiddlewareRequestStreamPausing()
+    {
+        $loop = Factory::create();
+        $connector = new Connector($loop);
+
+        $server = new StreamingServer(new MiddlewareRunner(array(
+            new LimitHandlersMiddleware(5),
+            new RequestBodyBufferMiddleware(16 * 1024 * 1024), // 16 MiB
+            function (ServerRequestInterface $request, $next) use ($loop) {
+                return new Promise\Promise(function ($resolve) use ($request, $loop, $next) {
+                    $loop->addTimer(0.1, function () use ($request, $resolve, $next) {
+                        $resolve($next($request));
+                    });
+                });
+            },
+            function (ServerRequestInterface $request) {
+                return new Response(200, array(), (string)strlen((string)$request->getBody()));
+            }
+        )));
+
+        $socket = new Socket(0, $loop);
+        $server->listen($socket);
+
+        $result = array();
+        for ($i = 0; $i < 6; $i++) {
+            $result[] = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) {
+                $conn->write(
+                    "GET / HTTP/1.0\r\nContent-Length: 1024\r\nHost: " . noScheme($conn->getRemoteAddress()) . "\r\n\r\n" .
+                    str_repeat('a', 1024) .
+                    "\r\n\r\n"
+                );
+
+                return Stream\buffer($conn);
+            });
+        }
+
+        $responses = Block\await(Promise\all($result), $loop, 1.0);
+
+        foreach ($responses as $response) {
+            $this->assertContains("HTTP/1.0 200 OK", $response, $response);
+            $this->assertTrue(substr($response, -4) == 1024, $response);
+        }
+
+        $socket->close();
+    }
+
 }
 
 function noScheme($uri)