Skeleton for LimitHandlersMiddleware

Author: WyriHaximus

src/Middleware/LimitHandlersMiddleware.php

@@ -0,0 +1,74 @@
+<?php
+
+namespace React\Http\Middleware;
+
+use Psr\Http\Message\ServerRequestInterface;
+use React\Promise\Deferred;
+use React\Stream\ReadableStreamInterface;
+use SplQueue;
+
+final class LimitHandlersMiddleware
+{
+    private $limit;
+    private $pending = 0;
+    private $queued;
+
+    /**
+     * @param int $limit Maximum amount of concurrent requests handled.
+     *
+     * For example when $limit is set to 10, 10 requests will flow to $next
+     * while more incoming requests have to wait until one is done.
+     */
+    public function __construct($limit)
+    {
+        $this->limit = $limit;
+        $this->queued = new SplQueue();
+    }
+
+    public function __invoke(ServerRequestInterface $request, $next)
+    {
+        $body = $request->getBody();
+        if ($body instanceof ReadableStreamInterface) {
+            $body->pause();
+        }
+        $deferred = new Deferred();
+        $this->queued->enqueue($deferred);
+
+        $this->processQueue();
+
+        $that = $this;
+        $pending = &$this->pending;
+        return $deferred->promise()->then(function () use ($request, $next, &$pending) {
+            $pending++;
+            $body = $request->getBody();
+            if ($body instanceof ReadableStreamInterface) {
+                $body->resume();
+            }
+            return $next($request);
+        })->then(function ($response) use ($that, &$pending) {
+            $pending--;
+            $that->processQueue();
+            return $response;
+        }, function ($error) use ($that, &$pending) {
+            $pending--;
+            $that->processQueue();
+            return $error;
+        });
+    }
+
+    /**
+     * @internal
+     */
+    public function processQueue()
+    {
+        if ($this->pending >= $this->limit) {
+            return;
+        }
+
+        if ($this->queued->count() === 0) {
+            return;
+        }
+
+        $this->queued->dequeue()->resolve();
+    }
+}

tests/Middleware/LimitHandlersMiddlewareTest.php

@@ -0,0 +1,100 @@
+<?php
+
+namespace React\Tests\Http\Middleware;
+
+use React\Http\Middleware\LimitHandlersMiddleware;
+use React\Http\Io\ServerRequest;
+use React\Promise\Deferred;
+use React\Tests\Http\TestCase;
+
+final class LimitHandlersMiddlewareTest extends TestCase
+{
+    public function testLimitOneRequestConcurrently()
+    {
+        /**
+         * The first request
+         */
+        $requestA = new ServerRequest('GET', 'https://example.com/');
+        $deferredA = new Deferred();
+        $calledA = false;
+        $nextA = function () use (&$calledA, $deferredA) {
+            $calledA = true;
+            return $deferredA->promise();
+        };
+
+        /**
+         * The second request
+         */
+        $requestB = new ServerRequest('GET', 'https://www.example.com/');
+        $deferredB = new Deferred();
+        $calledB = false;
+        $nextB = function () use (&$calledB, $deferredB) {
+            $calledB = true;
+            return $deferredB->promise();
+        };
+
+        /**
+         * The third request
+         */
+        $requestC = new ServerRequest('GET', 'https://www.example.com/');
+        $calledC = false;
+        $nextC = function () use (&$calledC) {
+            $calledC = true;
+        };
+
+        /**
+         * The handler
+         *
+         */
+        $limitHandlers = new LimitHandlersMiddleware(1);
+
+        $this->assertFalse($calledA);
+        $this->assertFalse($calledB);
+        $this->assertFalse($calledC);
+
+        $limitHandlers($requestA, $nextA);
+
+        $this->assertTrue($calledA);
+        $this->assertFalse($calledB);
+        $this->assertFalse($calledC);
+
+        $limitHandlers($requestB, $nextB);
+
+        $this->assertTrue($calledA);
+        $this->assertFalse($calledB);
+        $this->assertFalse($calledC);
+
+        $limitHandlers($requestC, $nextC);
+
+        $this->assertTrue($calledA);
+        $this->assertFalse($calledB);
+        $this->assertFalse($calledC);
+
+        /**
+         * Ensure resolve frees up a slot
+         */
+        $deferredA->resolve();
+
+        $this->assertTrue($calledA);
+        $this->assertTrue($calledB);
+        $this->assertFalse($calledC);
+
+        /**
+         * Ensure reject also frees up a slot
+         */
+        $deferredB->reject();
+
+        $this->assertTrue($calledA);
+        $this->assertTrue($calledB);
+        $this->assertTrue($calledC);
+    }
+
+    public function testStreamPauseAndResume()
+    {
+        $body = $this->getMockBuilder('React\Http\Io\HttpBodyStream')->disableOriginalConstructor()->getMock();
+        $body->expects($this->once())->method('pause');
+        $body->expects($this->once())->method('resume');
+        $limitHandlers = new LimitHandlersMiddleware(1);
+        $limitHandlers(new ServerRequest('GET', 'https://example.com/', array(), $body), function () {});
+    }
+}