Middleware

README.md

@@ -9,6 +9,7 @@ Event-driven, streaming plaintext HTTP and secure HTTPS server for [ReactPHP](ht
 * [Quickstart example](#quickstart-example)
 * [Usage](#usage)
   * [Server](#server)
+  * [Middleware](#middleware)
   * [Request](#request)
   * [Response](#response)
 * [Install](#install)
@@ -136,6 +137,88 @@ $server->on('error', function (Exception $e) {
 Note that the request object can also emit an error.
 Check out [request](#request) for more details.
 
+### Middleware
+
+As of `0.8` request handling is done via middlewares either by passing
+a `callable` as before, an array with middlewares implementing 
+[`React\Http\MiddlewareInterface`](src/MiddlewareInterface.php),
+passing a concrete implementation of 
+[`React\Http\MiddlewareStackInterface`](src/MiddlewareStackInterface.php)
+directly.
+
+Middlewares implementing [`React\Http\MiddlewareInterface`](src/MiddlewareInterface.php) 
+are expected to return a promise resolving a 
+`Psr\Http\Message\ResponseInterface`.
+However what ever comes out of the middleware is run through 
+[`resolve()`](https://github.com/reactphp/promise#resolve)
+and turned into a promise regardless. When a middleware can't resolve the 
+request to a promise it is expected to delegate that task to the passed 
+[`React\Http\MiddlewareStackInterface`](src/MiddlewareStackInterface.php).
+This is process is inspired by the work in progress PSR-15 with the 
+difference that promises are returned instead of responses.
+
+See also [example #12](examples) and [example #13](examples).
+
+#### Callback
+
+A few build in middlewares are shipped with this package, one of them is
+the `Callback` middleware. This middlewares takes a `callable` and executes 
+it on each request. Identically to how this package worked before `0.8`.
+
+Usage:
+
+```php
+$middlewares = [
+    new Callback(function (ServerRequestInterface $request) {
+        return new Response(200);
+    }),
+];
+```
+
+#### Buffer
+
+Another build in middleware is `Buffer` which will buffer the incoming 
+request body until the reported size has been reached. Then it will 
+call the middleware stack with the new request instance containing the 
+full request body.
+
+Usage:
+
+```php
+$middlewares = [
+    new Buffer(),
+];
+```
+
+#### LimitHandlers
+
+The `LimitHandlers` middleware is all about concurrency control and will only allow the
+given number of requests to be handled concurrently.
+
+Usage:
+
+```php
+$middlewares = [
+    new LimitHandlers(10),
+];
+```
+
+#### Stacking middlewares
+
+All these middlewares can of course be combined into a stack. The following example limits 
+to 5 concurrent handlers, buffers the request body before handling and returns a empty 200 
+response.
+
+```php
+$middlewares = [
+    new LimitHandlers(5),
+    new Buffer(),
+    new Callback(function (ServerRequestInterface $request) {
+        return new Response(200);
+    }),
+];
+```
+
 ### Request
 
 An seen above, the `Server` class is responsible for handling incoming
@@ -659,7 +742,7 @@ The recommended way to install this library is [through Composer](http://getcomp
 This will install the latest supported version:
 
 ```bash
-$ composer require react/http:^0.7.3
+$ composer require react/http:^0.7.4
 ```
 
 More details about version upgrades can be found in the [CHANGELOG](CHANGELOG.md).

composer.json

@@ -9,7 +9,8 @@
         "react/socket": "^1.0 || ^0.8 || ^0.7 || ^0.6 || ^0.5",
         "react/stream": "^1.0 || ^0.7 || ^0.6 || ^0.5 || ^0.4.6",
         "react/promise": "^2.3 || ^1.2.1",
-        "evenement/evenement": "^3.0 || ^2.0 || ^1.0"
+        "evenement/evenement": "^3.0 || ^2.0 || ^1.0",
+        "react/promise-stream": "^0.1.1"
     },
     "autoload": {
         "psr-4": {
@@ -18,7 +19,6 @@
     },
     "require-dev": {
         "phpunit/phpunit": "^4.8.10||^5.0",
-        "react/promise-stream": "^0.1.1",
         "react/socket": "^1.0 || ^0.8 || ^0.7",
         "clue/block-react": "^1.1"
     }

examples/12-middleware.php

@@ -0,0 +1,42 @@
+<?php
+
+use Psr\Http\Message\ServerRequestInterface;
+use React\EventLoop\Factory;
+use React\Http\Middleware\Buffer;
+use React\Http\Middleware\LimitHandlers;
+use React\Http\MiddlewareRunner;
+use React\Http\Response;
+use React\Http\Server;
+use React\Promise\Deferred;
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$loop = Factory::create();
+
+$server = new Server(new MiddlewareRunner(
+    new Response(404),
+    array(
+        new LimitHandlers(3), // Only handle three requests concurrently
+        new Buffer(), // Buffer the whole request body before proceeding
+        function (ServerRequestInterface $request) use ($loop) {
+            $deferred = new Deferred();
+            $loop->futureTick(function () use ($deferred) {
+                $deferred->resolve(new Response(
+                    200,
+                    array(
+                        'Content-Type' => 'text/plain'
+                    ),
+                    "Hello world\n"
+                ));
+            });
+            return $deferred->promise();
+        },
+    )
+));
+
+$socket = new \React\Socket\Server(isset($argv[1]) ? $argv[1] : '0.0.0.0:0', $loop);
+$server->listen($socket);
+
+echo 'Listening on ' . str_replace('tcp:', 'http:', $socket->getAddress()) . PHP_EOL;
+
+$loop->run();

examples/13-request-sec.php

@@ -0,0 +1,69 @@
+<?php
+
+use Psr\Http\Message\ServerRequestInterface;
+use React\EventLoop\Factory;
+use React\Http\MiddlewareRunner;
+use React\Http\Response;
+use React\Http\Server;
+use React\Promise\Deferred;
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$total = 0;
+$counts = array(
+    'requests'  => 0,
+    'responses' => 0,
+);
+final class Incre
+{
+    public function __invoke(ServerRequestInterface $request, callable $stack)
+    {
+        global $counts, $total;
+        $total++;
+        $counts['requests']++;
+        return $stack($request)->then(function ($response) {
+            global $counts;
+            $counts['responses']++;
+            return $response;
+        });
+    }
+}
+
+$loop = Factory::create();
+
+$loop->addPeriodicTimer(1, function () use (&$counts, &$total) {
+    echo 'Req/s:  ', number_format($counts['requests']), PHP_EOL;
+    echo 'Resp/s: ', number_format($counts['responses']), PHP_EOL;
+    echo 'Total:  ', number_format($total), PHP_EOL;
+    echo '---------------------', PHP_EOL;
+    $counts = array(
+        'requests'  => 0,
+        'responses' => 0,
+    );
+});
+$server = new Server(new MiddlewareRunner(
+    new Response(500),
+    array(
+        new Incre($counts),
+        function (ServerRequestInterface $request) use ($loop) {
+            $deferred = new Deferred();
+            $loop->addTimer(mt_rand(1, 10) / 10, function () use ($deferred) {
+                $deferred->resolve(new Response(
+                    200,
+                    array(
+                        'Content-Type' => 'text/plain'
+                    ),
+                    "Hello world\n"
+                ));
+            });
+            return $deferred->promise();
+        }
+    ))
+);
+
+$socket = new \React\Socket\Server(isset($argv[1]) ? $argv[1] : '0.0.0.0:0', $loop);
+$server->listen($socket);
+
+echo 'Listening on ' . str_replace('tcp:', 'http:', $socket->getAddress()) . PHP_EOL;
+
+$loop->run();

src/Middleware/Buffer.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace React\Http\Middleware;
+
+use Psr\Http\Message\ServerRequestInterface;
+use React\Http\Response;
+use React\Promise\Stream;
+use React\Stream\ReadableStreamInterface;
+use RingCentral\Psr7\BufferStream;
+
+final class Buffer
+{
+    private $sizeLimit;
+
+    public function __construct($sizeLimit = null)
+    {
+        if ($sizeLimit === null) {
+            $sizeLimit = $this->iniMaxPostSize();
+        }
+
+        $this->sizeLimit = $sizeLimit;
+    }
+
+    public function __invoke(ServerRequestInterface $request, callable $stack)
+    {
+        $size = $request->getBody()->getSize();
+
+        if ($size === null) {
+            return new Response(411, array('Content-Type' => 'text/plain'), 'No Content-Length header given');
+        }
+
+        if ($size > $this->sizeLimit) {
+            return new Response(413, array('Content-Type' => 'text/plain'), 'Request body exceeds allowed limit');
+        }
+
+        $body = $request->getBody();
+        if (!$body instanceof ReadableStreamInterface) {
+            return $stack($request);
+        }
+
+        return Stream\buffer($body)->then(function ($buffer) use ($request, $stack) {
+            $stream = new BufferStream(strlen($buffer));
+            $stream->write($buffer);
+            $request = $request->withBody($stream);
+
+            return $stack($request);
+        });
+    }
+
+    private function iniMaxPostSize()
+    {
+        $size = ini_get('post_max_size');
+        $suffix = strtoupper(substr($size, -1));
+        if ($suffix === 'K') {
+            return substr($size, 0, -1) * 1024;
+        }
+        if ($suffix === 'M') {
+            return substr($size, 0, -1) * 1024 * 1024;
+        }
+        if ($suffix === 'G') {
+            return substr($size, 0, -1) * 1024  * 1024 * 1024;
+        }
+
+        return $size;
+    }
+}

src/Middleware/LimitHandlers.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace React\Http\Middleware;
+
+use Psr\Http\Message\ServerRequestInterface;
+use React\Http\MiddlewareInterface;
+use React\Http\MiddlewareStackInterface;
+use React\Promise\Deferred;
+
+final class LimitHandlers
+{
+    private $limit = 10;
+    private $pending = 0;
+    private $queued;
+
+    public function __construct($limit = 10)
+    {
+        $this->limit = $limit;
+        $this->queued = new \SplQueue();
+    }
+
+    public function __invoke(ServerRequestInterface $request, callable $stack)
+    {
+        $deferred = new Deferred();
+        $this->queued->enqueue($deferred);
+
+        $this->processQueue();
+
+        return $deferred->promise()->then(function () use ($request, $stack) {
+            $this->pending++;
+            return $stack($request);
+        })->then(function ($response) {
+            $this->pending--;
+            $this->processQueue();
+            return $response;
+        });
+    }
+
+    private function processQueue()
+    {
+        if ($this->pending >= $this->limit) {
+            return;
+        }
+
+        if ($this->queued->count() === 0) {
+            return;
+        }
+
+        $this->queued->dequeue()->resolve();
+    }
+}