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();
+ }
+}