Added ability to cancel await

examples/cancelAwait.php

@@ -0,0 +1,16 @@
+<?php
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$source = \Rx\Observable::interval(1000);
+
+$generator = \Rx\await($source);
+
+foreach ($generator as $item) {
+    if ($item === 3) {
+        \Rx\cancelAwait($generator);
+    }
+    echo $item, PHP_EOL;
+}
+
+echo 'DONE';

src/functions.php

@@ -3,7 +3,8 @@
 namespace Rx;
 
 use React\EventLoop\LoopInterface;
-use Rx\Observer\CallbackObserver;
+
+const AWAIT_CANCEL = 'cancel';
 
 /**
  * Wait until observable completes.
@@ -12,13 +13,13 @@
  * @param LoopInterface $loop
  * @return \Generator
  */
-function await(Observable $observable, LoopInterface $loop = null)
+function await(Observable $observable, LoopInterface $loop = null): \Generator
 {
     $completed = false;
     $results   = [];
     $loop      = $loop ?: \EventLoop\getLoop();
 
-    $observable->subscribe(new CallbackObserver(
+    $disposable = $observable->subscribe(
         function ($value) use (&$results, $loop) {
             $results[] = $value;
 
@@ -30,18 +31,26 @@ function ($e) use (&$completed) {
         },
         function () use (&$completed) {
             $completed = true;
-        }
-
-    ));
+        });
 
     while (!$completed) {
 
         $loop->run();
 
         foreach ($results as $result) {
-            yield $result;
+            $cancel = yield $result;
+
+            if ($cancel === AWAIT_CANCEL) {
+                $disposable->dispose();
+                return;
+            }
         }
 
         $results = [];
     }
 }
+
+function cancelAwait(\Generator $generator)
+{
+    $generator->send(AWAIT_CANCEL);
+}

tests/FunctionAwaitTest.php

@@ -2,6 +2,7 @@
 
 namespace Rx\React\Tests;
 
+use function Rx\cancelAwait;
 use Rx\Functional\FunctionalTestCase;
 use Rx\Observable;
 
@@ -68,4 +69,28 @@ public function await_default_timeout()
 
     }
 
+    /**
+     * @test
+     */
+    public function await_cancel()
+    {
+        $array = [1, 2, 3];
+
+        $observable = Observable::fromArray($array);
+
+        $generator = \Rx\await($observable);
+
+        $result = [];
+        foreach ($generator as $item) {
+            if ($item === 2) {
+                cancelAwait($generator);
+            }
+
+            $result[] = $item;
+        }
+
+        $this->assertEquals([1, 2], $result);
+        $this->assertEquals(count($result), 2);
+
+    }
 }