Feature/add promisesraces cancel mechanism

examples/05-cancellable-promise.php

@@ -0,0 +1,57 @@
+#!/usr/bin/env php
+<?php
+
+require __DIR__.'/../vendor/autoload.php';
+
+use M6Web\Tornado\Adapter;
+use M6Web\Tornado\EventLoop;
+
+// Choose your adapter.
+$eventLoop = new Adapter\Tornado\EventLoop();
+//$eventLoop = new Adapter\Tornado\SynchronousEventLoop();
+//$eventLoop = new Adapter\Amp\EventLoop();
+//$eventLoop = new Adapter\ReactPhp\EventLoop(new React\EventLoop\StreamSelectLoop());
+
+function timer(EventLoop $eventLoop, string $id, int $time, \M6Web\Tornado\Promise &$promise = null)
+{
+    $result = 'not resolved';
+    try {
+        echo "[$id] Starting to wait $time …\n";
+        yield $eventLoop->delay($time);
+        echo "[$id] One more time !\n";
+        yield $eventLoop->delay($time);
+
+        if ($promise) {
+            echo "[$id] Cancelling other promise …\n";
+            $promise->cancel();
+            echo "[$id] Done!\n";
+        }
+
+        $result = $id;
+    } catch (\Exception $e) {
+        echo "[$id] XXX Cancelled …\n";
+    }
+
+    return $result;
+}
+
+echo "Let's start!\n";
+
+try {
+    $result = $eventLoop->wait(
+        $eventLoop->promiseAll(
+            $p1 = $eventLoop->async(timer($eventLoop, 'A', 100)),
+            $eventLoop->async(timer($eventLoop, 'B', 10, $p1))
+        )
+    );
+} catch (\M6Web\Tornado\CancelledException $e) {
+    $result = 'cancelled promise';
+} catch (\Exception $e) {
+    $result = 'other exception';
+}
+
+echo "async cancellation result :\n";
+
+var_dump($result);
+
+echo "Finished!\n";

examples/06-advanced-cancellable-promise.php

@@ -0,0 +1,73 @@
+#!/usr/bin/env php
+<?php
+
+require __DIR__.'/../vendor/autoload.php';
+
+use M6Web\Tornado\Adapter;
+use M6Web\Tornado\EventLoop;
+
+// Choose your adapter.
+$eventLoop = new Adapter\Tornado\EventLoop();
+//$eventLoop = new Adapter\Tornado\SynchronousEventLoop();
+//$eventLoop = new Adapter\Amp\EventLoop();
+//$eventLoop = new Adapter\ReactPhp\EventLoop(new React\EventLoop\StreamSelectLoop());
+
+function timer(EventLoop $eventLoop, string $id, int $time, \M6Web\Tornado\Promise &$promise = null)
+{
+    $result = 'not resolved';
+    try {
+        echo "[$id] Starting to wait $time …\n";
+        yield $eventLoop->delay($time);
+        echo "[$id] Working\n";
+        yield $eventLoop->delay($time);
+        echo "[$id] Working\n";
+        yield $eventLoop->delay($time);
+
+        if ($promise) {
+            echo "[$id] Cancelling other promise …\n";
+            $promise->cancel();
+            echo "[$id] Cancellation Done!\n";
+        }
+
+        echo "[$id] Working after cancellation !\n";
+        yield $eventLoop->delay($time);
+        echo "[$id] Working after cancellation !\n";
+        yield $eventLoop->delay($time);
+        echo "[$id] Working after cancellation !\n";
+        yield $eventLoop->delay($time);
+        echo "[$id] Working after cancellation !\n";
+        yield $eventLoop->delay($time);
+        echo "[$id] Working after cancellation !\n";
+        yield $eventLoop->delay($time);
+
+        $result = $id;
+    } catch (\Exception $e) {
+        echo "[$id] Cancelled … : clean workspace and other stuff\n";
+    }
+
+    return $result;
+}
+
+echo "Let's start!\n";
+
+try {
+    $result = $eventLoop->wait(
+        $eventLoop->promiseAll(
+            $promise = $eventLoop->promiseAll(
+                $eventLoop->async(timer($eventLoop, ' Timer A ', 100)),
+                $eventLoop->async(timer($eventLoop, ' Timer B ', 100))
+            ),
+            $eventLoop->async(timer($eventLoop, 'Canceller', 100, $promise))
+        )
+    );
+} catch (\M6Web\Tornado\CancelledException $e) {
+    $result = 'cancelled promise';
+} catch (\Exception $e) {
+    $result = 'other exception';
+}
+
+echo "async cancellation with result conservation:\n";
+
+var_dump($result);
+
+echo "Finished!\n";

phpunit.xml.dist

@@ -22,6 +22,24 @@
         <testsuite name="Tornado Examples">
             <directory>examples/tests</directory>
         </testsuite>
+        <testsuite name="Amp">
+            <directory>/usr/local/var/www/Tornado/tests/Adapter/Amp/EventLoopTest.php</directory>
+        </testsuite>
+        <testsuite name="Guzzle">
+            <directory>/usr/local/var/www/Tornado/tests/Adapter/Guzzle/HttpClientTest.php</directory>
+        </testsuite>
+        <testsuite name="ReactPhp">
+            <directory>/usr/local/var/www/Tornado/tests/Adapter/ReactPhp/EventLoopTest.php</directory>
+        </testsuite>
+        <testsuite name="Symfony">
+            <directory>/usr/local/var/www/Tornado/tests/Adapter/Symfony/HttpClientTest.php</directory>
+        </testsuite>
+        <testsuite name="Tornado">
+            <file>/usr/local/var/www/Tornado/tests/Adapter/Tornado/EventLoopTest.php</file>
+        </testsuite>
+        <testsuite name="Tornado Synchronous">
+            <file>/usr/local/var/www/Tornado/tests/Adapter/Tornado/SynchronousEventLoopTest.php</file>
+        </testsuite>
     </testsuites>
 
     <filter>

src/Adapter/Amp/EventLoop.php

@@ -3,6 +3,7 @@
 namespace M6Web\Tornado\Adapter\Amp;
 
 use M6Web\Tornado\Adapter\Common;
+use M6Web\Tornado\CancelledException;
 use M6Web\Tornado\Deferred;
 use M6Web\Tornado\Promise;
 
@@ -41,13 +42,17 @@ public function wait(Promise $promise)
      */
     public function async(\Generator $generator): Promise
     {
-        $wrapper = function (\Generator $generator, \Amp\Deferred $deferred): \Generator {
+        /** @var Promise $currentPromise */
+        $currentPromise = null;
+
+        $wrapper = function (\Generator $generator, \Amp\Deferred $deferred) use (&$currentPromise): \Generator {
             try {
                 while ($generator->valid()) {
                     $blockingPromise = $generator->current();
                     if (!$blockingPromise instanceof Promise) {
                         throw new \Error('Asynchronous function is yielding a ['.gettype($blockingPromise).'] instead of a Promise.');
                     }
+                    $currentPromise = $blockingPromise;
                     $blockingPromise = Internal\PromiseWrapper::toHandledPromise(
                         $blockingPromise,
                         $this->unhandledFailingPromises
@@ -79,14 +84,24 @@ public function async(\Generator $generator): Promise
         $deferred = new \Amp\Deferred();
         \Amp\Promise\rethrow(new \Amp\Coroutine($wrapper($generator, $deferred)));
 
-        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises);
+        $cancellable = function () use (&$currentPromise) {
+            $currentPromise->cancel();
+        };
+
+        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises, $cancellable);
     }
 
     /**
      * {@inheritdoc}
      */
     public function promiseAll(Promise ...$promises): Promise
     {
+        $cancellable = function () use (&$promises) {
+            foreach ($promises as $promise) {
+                $promise->cancel();
+            }
+        };
+
         return Internal\PromiseWrapper::createUnhandled(
             \Amp\Promise\all(
                 array_map(
@@ -99,7 +114,8 @@ function (Promise $promise) {
                     $promises
                 )
             ),
-            $this->unhandledFailingPromises
+            $this->unhandledFailingPromises,
+            $cancellable
         );
     }
 
@@ -121,6 +137,8 @@ public function promiseForeach($traversable, callable $function): Promise
      */
     public function promiseRace(Promise ...$promises): Promise
     {
+        $toWrapPromises = [];
+        $promisesCancellation = null;
         if (empty($promises)) {
             return $this->promiseFulfilled(null);
         }
@@ -144,28 +162,45 @@ public function promiseRace(Promise ...$promises): Promise
         };
 
         $promises = array_map(
-            function (Promise $promise) {
-                return Internal\PromiseWrapper::toHandledPromise(
+            function (Promise $promise) use (&$toWrapPromises) {
+                $tempPromise = Internal\PromiseWrapper::toHandledPromise(
                     $promise,
                     $this->unhandledFailingPromises
-                )->getAmpPromise();
+                );
+                $toWrapPromises[] = $tempPromise;
+
+                return $tempPromise->getAmpPromise();
             },
             $promises
         );
 
         foreach ($promises as $index => $promise) {
             \Amp\Promise\rethrow(new \Amp\Coroutine($wrapPromise($promise)));
         }
+        $promisesCancellation = function () use (&$toWrapPromises) {
+            foreach ($toWrapPromises as $index => $promise) {
+                $promise->cancel();
+            }
+        };
+
+        $cancellation = function () use (&$deferred, &$promisesCancellation) {
+            $deferred->fail(new CancelledException('promise race cancellation'));
+            ($promisesCancellation)();
+        };
 
-        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises);
+        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises, $cancellation);
     }
 
     /**
      * {@inheritdoc}
      */
     public function promiseFulfilled($value): Promise
     {
-        return Internal\PromiseWrapper::createHandled(new \Amp\Success($value));
+        return Internal\PromiseWrapper::createHandled(
+            new \Amp\Success($value),
+            function () {
+            }
+        );
     }
 
     /**
@@ -174,7 +209,11 @@ public function promiseFulfilled($value): Promise
     public function promiseRejected(\Throwable $throwable): Promise
     {
         // Manually created promises are considered as handled.
-        return Internal\PromiseWrapper::createHandled(new \Amp\Failure($throwable));
+        return Internal\PromiseWrapper::createHandled(
+            new \Amp\Failure($throwable),
+            function () {
+            }
+        );
     }
 
     /**
@@ -184,11 +223,16 @@ public function idle(): Promise
     {
         $deferred = new \Amp\Deferred();
 
-        \Amp\Loop::defer(function () use ($deferred) {
+        $deferedId = \Amp\Loop::defer(function () use ($deferred) {
             $deferred->resolve();
         });
 
-        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises);
+        $cancellation = function () use ($deferedId, $deferred) {
+            \Amp\Loop::cancel($deferedId);
+            $deferred->fail(new CancelledException('Delay cancelled'));
+        };
+
+        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises, $cancellation);
     }
 
     /**
@@ -198,22 +242,27 @@ public function delay(int $milliseconds): Promise
     {
         $deferred = new \Amp\Deferred();
 
-        \Amp\Loop::delay($milliseconds, function () use ($deferred) {
+        $delayId = \Amp\Loop::delay($milliseconds, function () use ($deferred) {
             $deferred->resolve();
         });
 
-        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises);
+        $cancellation = function () use ($delayId, $deferred) {
+            \Amp\Loop::cancel($delayId);
+            $deferred->fail(new CancelledException('Delay cancelled'));
+        };
+
+        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises, $cancellation);
     }
 
     /**
      * {@inheritdoc}
      */
-    public function deferred(): Deferred
+    public function deferred(callable $canceller = null): Deferred
     {
         return new Internal\Deferred(
             $deferred = new \Amp\Deferred(),
             // Manually created promises are considered as handled.
-            Internal\PromiseWrapper::createHandled($deferred->promise())
+            Internal\PromiseWrapper::createHandled($deferred->promise(), function () {})
         );
     }
 
@@ -232,7 +281,9 @@ function ($watcherId, $stream) use ($deferred) {
             }
         );
 
-        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises);
+        $cancellation = function () {};
+
+        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises, $cancellation);
     }
 
     /**
@@ -250,7 +301,9 @@ function ($watcherId, $stream) use ($deferred) {
             }
         );
 
-        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises);
+        $cancellation = function () {};
+
+        return Internal\PromiseWrapper::createUnhandled($deferred->promise(), $this->unhandledFailingPromises, $cancellation);
     }
 
     public function __construct()