Support promise cancellation for coroutine() and clean up garbage references

src/functions.php

@@ -224,22 +224,33 @@ function coroutine(callable $function, ...$args): PromiseInterface
         return resolve($generator);
     }
 
-    $deferred = new Deferred();
+    $promise = null;
+    $deferred = new Deferred(function () use (&$promise) {
+        // cancel pending promise(s) as long as generator function keeps yielding
+        while ($promise instanceof CancellablePromiseInterface) {
+            $temp = $promise;
+            $promise = null;
+            $temp->cancel();
+        }
+    });
 
     /** @var callable $next */
-    $next = function () use ($deferred, $generator, &$next) {
+    $next = function () use ($deferred, $generator, &$next, &$promise) {
         try {
             if (!$generator->valid()) {
+                $next = null;
                 $deferred->resolve($generator->getReturn());
                 return;
             }
         } catch (\Throwable $e) {
+            $next = null;
             $deferred->reject($e);
             return;
         }
 
         $promise = $generator->current();
         if (!$promise instanceof PromiseInterface) {
+            $next = null;
             $deferred->reject(new \UnexpectedValueException(
                 'Expected coroutine to yield ' . PromiseInterface::class . ', but got ' . (is_object($promise) ? get_class($promise) : gettype($promise))
             ));
@@ -252,7 +263,8 @@ function coroutine(callable $function, ...$args): PromiseInterface
         }, function (\Throwable $reason) use ($generator, $next) {
             $generator->throw($reason);
             $next();
-        })->then(null, function (\Throwable $reason) use ($deferred) {
+        })->then(null, function (\Throwable $reason) use ($deferred, &$next) {
+            $next = null;
             $deferred->reject($reason);
         });
     };

tests/CoroutineTest.php

@@ -2,6 +2,7 @@
 
 namespace React\Tests\Async;
 
+use React\Promise\Promise;
 use function React\Async\coroutine;
 use function React\Promise\reject;
 use function React\Promise\resolve;
@@ -104,4 +105,137 @@ public function testCoroutineReturnsRejectedPromiseIfFunctionYieldsInvalidValue(
 
         $promise->then(null, $this->expectCallableOnceWith(new \UnexpectedValueException('Expected coroutine to yield React\Promise\PromiseInterface, but got integer')));
     }
+
+
+    public function testCoroutineWillCancelPendingPromiseWhenCallingCancelOnResultingPromise()
+    {
+        $cancelled = 0;
+        $promise = coroutine(function () use (&$cancelled) {
+            yield new Promise(function () use (&$cancelled) {
+                ++$cancelled;
+            });
+        });
+
+        $promise->cancel();
+
+        $this->assertEquals(1, $cancelled);
+    }
+
+    public function testCoroutineWillCancelAllPendingPromisesWhenFunctionContinuesToYieldWhenCallingCancelOnResultingPromise()
+    {
+        $promise = coroutine(function () {
+            $promise = new Promise(function () { }, function () {
+                throw new \RuntimeException('Frist operation cancelled', 21);
+            });
+
+            try {
+                yield $promise;
+            } catch (\RuntimeException $e) {
+                // ignore exception and continue
+            }
+
+            yield new Promise(function () { }, function () {
+                throw new \RuntimeException('Second operation cancelled', 42);
+            });
+        });
+
+        $promise->cancel();
+
+        $promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Second operation cancelled', 42)));
+    }
+
+    public function testCoroutineShouldNotCreateAnyGarbageReferencesWhenGeneratorReturns()
+    {
+        if (class_exists('React\Promise\When')) {
+            $this->markTestSkipped('Not supported on legacy Promise v1 API');
+        }
+
+        gc_collect_cycles();
+        gc_collect_cycles();
+
+        $promise = coroutine(function () {
+            if (false) {
+                yield;
+            }
+            return 42;
+        });
+
+        unset($promise);
+
+        $this->assertEquals(0, gc_collect_cycles());
+    }
+
+    public function testCoroutineShouldNotCreateAnyGarbageReferencesForPromiseRejectedWithExceptionImmediately()
+    {
+        if (class_exists('React\Promise\When')) {
+            $this->markTestSkipped('Not supported on legacy Promise v1 API');
+        }
+
+        gc_collect_cycles();
+
+        $promise = coroutine(function () {
+            yield new Promise(function () {
+                throw new \RuntimeException('Failed', 42);
+            });
+        });
+
+        unset($promise);
+
+        $this->assertEquals(0, gc_collect_cycles());
+    }
+
+    public function testCoroutineShouldNotCreateAnyGarbageReferencesForPromiseRejectedWithExceptionOnCancellation()
+    {
+        if (class_exists('React\Promise\When')) {
+            $this->markTestSkipped('Not supported on legacy Promise v1 API');
+        }
+
+        gc_collect_cycles();
+
+        $promise = coroutine(function () {
+            yield new Promise(function () { }, function () {
+                throw new \RuntimeException('Operation cancelled', 42);
+            });
+        });
+
+        $promise->cancel();
+        unset($promise);
+
+        $this->assertEquals(0, gc_collect_cycles());
+    }
+
+    public function testCoroutineShouldNotCreateAnyGarbageReferencesWhenGeneratorThrowsBeforeFirstYield()
+    {
+        if (class_exists('React\Promise\When')) {
+            $this->markTestSkipped('Not supported on legacy Promise v1 API');
+        }
+
+        gc_collect_cycles();
+
+        $promise = coroutine(function () {
+            throw new \RuntimeException('Failed', 42);
+            yield;
+        });
+
+        unset($promise);
+
+        $this->assertEquals(0, gc_collect_cycles());
+    }
+
+    public function testCoroutineShouldNotCreateAnyGarbageReferencesWhenGeneratorYieldsInvalidValue()
+    {
+        if (class_exists('React\Promise\When')) {
+            $this->markTestSkipped('Not supported on legacy Promise v1 API');
+        }
+
+        gc_collect_cycles();
+
+        $promise = coroutine(function () {
+            yield 42;
+        });
+
+        unset($promise);
+
+        $this->assertEquals(0, gc_collect_cycles());
+    }
 }