Merge pull request #293 from clue-labs/parallel-next

Fix concurrent next middleware request handlers and recurse next request handlers

Author: WyriHaximus

src/Io/MiddlewareRunner.php

@@ -16,9 +16,8 @@ final class MiddlewareRunner
 {
     /**
      * @var callable[]
-     * @internal
      */
-    public $middleware = array();
+    private $middleware = array();
 
     /**
      * @param callable[] $middleware
@@ -38,34 +37,26 @@ public function __invoke(ServerRequestInterface $request)
             return Promise\reject(new \RuntimeException('No middleware to run'));
         }
 
-        $position = 0;
+        return $this->call($request, 0);
+    }
 
+    /** @internal */
+    public function call(ServerRequestInterface $request, $position)
+    {
         $that = $this;
-        $func = function (ServerRequestInterface $request) use (&$func, &$position, &$that) {
-            $middleware = $that->middleware[$position];
-            $response = null;
-            $promise = new Promise\Promise(function ($resolve) use ($middleware, $request, $func, &$response, &$position) {
-                $position++;
-
-                $response = $middleware(
-                    $request,
-                    $func
-                );
-
-                $resolve($response);
-            }, function () use (&$response) {
-                if ($response instanceof Promise\CancellablePromiseInterface) {
-                    $response->cancel();
-                }
-            });
-
-            return $promise->then(null, function ($error) use (&$position) {
-                $position--;
-
-                return Promise\reject($error);
-            });
+        $next = function (ServerRequestInterface $request) use ($that, $position) {
+            return $that->call($request, $position + 1);
         };
 
-        return $func($request);
+        $handler = $this->middleware[$position];
+        try {
+            return Promise\resolve($handler($request, $next));
+        } catch (\Exception $error) {
+            // request handler callback throws an Exception
+            return Promise\reject($error);
+        } catch (\Throwable $error) { // @codeCoverageIgnoreStart
+            // request handler callback throws a PHP7+ Error
+            return Promise\reject($error); // @codeCoverageIgnoreEnd
+        }
     }
 }

tests/Io/MiddlewareRunnerTest.php

@@ -12,6 +12,8 @@
 use React\Tests\Http\Middleware\ProcessStack;
 use React\Tests\Http\TestCase;
 use RingCentral\Psr7\Response;
+use Psr\Http\Message\RequestInterface;
+use React\Promise\CancellablePromiseInterface;
 
 final class MiddlewareRunnerTest extends TestCase
 {
@@ -28,6 +30,39 @@ public function testDefaultResponse()
         Block\await($middlewareStack($request), Factory::create());
     }
 
+    public function testReturnsRejectedPromiseIfHandlerThrowsException()
+    {
+        $middleware = new MiddlewareRunner(array(
+            function (ServerRequestInterface $request) {
+                throw new \RuntimeException('hello');
+            }
+        ));
+
+        $request = new ServerRequest('GET', 'http://example.com/');
+
+        $promise = $middleware($request);
+
+        $promise->then(null, $this->expectCallableOnceWith($this->isInstanceOf('RuntimeException')));
+    }
+
+    /**
+     * @requires PHP 7
+     */
+    public function testReturnsRejectedPromiseIfHandlerThrowsThrowable()
+    {
+        $middleware = new MiddlewareRunner(array(
+            function (ServerRequestInterface $request) {
+                throw new \Error('hello');
+            }
+        ));
+
+        $request = new ServerRequest('GET', 'http://example.com/');
+
+        $promise = $middleware($request);
+
+        $promise->then(null, $this->expectCallableOnceWith($this->isInstanceOf('Throwable')));
+    }
+
     public function provideProcessStackMiddlewares()
     {
         $processStackA = new ProcessStack();
@@ -342,4 +377,69 @@ public function testUncommonMiddlewareArrayFormats($middlewareFactory, $expected
 
         $this->assertSame($expectedSequence, (string) $response->getBody());
     }
+
+    public function testPendingNextRequestHandlersCanBeCalledConcurrently()
+    {
+        $called = 0;
+        $middleware = new MiddlewareRunner(array(
+            function (RequestInterface $request, $next) {
+                $first = $next($request);
+                $second = $next($request);
+
+                return new Response();
+            },
+            function (RequestInterface $request) use (&$called) {
+                ++$called;
+
+                return new Promise\Promise(function () { });
+            }
+        ));
+
+        $request = new ServerRequest('GET', 'http://example.com/');
+
+        $response = Block\await($middleware($request), Factory::create());
+
+        $this->assertTrue($response instanceof ResponseInterface);
+        $this->assertEquals(2, $called);
+    }
+
+    public function testCancelPendingNextHandler()
+    {
+        $once = $this->expectCallableOnce();
+        $middleware = new MiddlewareRunner(array(
+            function (RequestInterface $request, $next) {
+                $ret = $next($request);
+                $ret->cancel();
+
+                return $ret;
+            },
+            function (RequestInterface $request) use ($once) {
+                return new Promise\Promise(function () { }, $once);
+            }
+        ));
+
+        $request = new ServerRequest('GET', 'http://example.com/');
+
+        $middleware($request);
+    }
+
+    public function testCancelResultingPromiseWillCancelPendingNextHandler()
+    {
+        $once = $this->expectCallableOnce();
+        $middleware = new MiddlewareRunner(array(
+            function (RequestInterface $request, $next) {
+                return $next($request);
+            },
+            function (RequestInterface $request) use ($once) {
+                return new Promise\Promise(function () { }, $once);
+            }
+        ));
+
+        $request = new ServerRequest('GET', 'http://example.com/');
+
+        $promise = $middleware($request);
+
+        $this->assertTrue($promise instanceof CancellablePromiseInterface);
+        $promise->cancel();
+    }
 }