Take advantage of iterators instead of converting to array first

Author: clue

src/functions.php

@@ -548,19 +548,10 @@ function parallel(iterable $tasks): PromiseInterface
         $pending = [];
     });
     $results = [];
-    $errored = false;
+    $continue = true;
 
-    if (!\is_array($tasks)) {
-        $tasks = \iterator_to_array($tasks);
-    }
-
-    $numTasks = count($tasks);
-    if (0 === $numTasks) {
-        $deferred->resolve($results);
-    }
-
-    $taskErrback = function ($error) use (&$pending, $deferred, &$errored) {
-        $errored = true;
+    $taskErrback = function ($error) use (&$pending, $deferred, &$continue) {
+        $continue = false;
         $deferred->reject($error);
 
         foreach ($pending as $promise) {
@@ -572,25 +563,31 @@ function parallel(iterable $tasks): PromiseInterface
     };
 
     foreach ($tasks as $i => $task) {
-        $taskCallback = function ($result) use (&$results, &$pending, $numTasks, $i, $deferred) {
+        $taskCallback = function ($result) use (&$results, &$pending, &$continue, $i, $deferred) {
             $results[$i] = $result;
+            unset($pending[$i]);
 
-            if (count($results) === $numTasks) {
+            if (!$pending && !$continue) {
                 $deferred->resolve($results);
             }
         };
 
-        $promise = call_user_func($task);
+        $promise = \call_user_func($task);
         assert($promise instanceof PromiseInterface);
         $pending[$i] = $promise;
 
         $promise->then($taskCallback, $taskErrback);
 
-        if ($errored) {
+        if (!$continue) {
             break;
         }
     }
 
+    $continue = false;
+    if (!$pending) {
+        $deferred->resolve($results);
+    }
+
     return $deferred->promise();
 }
 
@@ -609,8 +606,9 @@ function series(iterable $tasks): PromiseInterface
     });
     $results = [];
 
-    if (!\is_array($tasks)) {
-        $tasks = \iterator_to_array($tasks);
+    if ($tasks instanceof \IteratorAggregate) {
+        $tasks = $tasks->getIterator();
+        assert($tasks instanceof \Iterator);
     }
 
     /** @var callable():void $next */
@@ -620,13 +618,19 @@ function series(iterable $tasks): PromiseInterface
     };
 
     $next = function () use (&$tasks, $taskCallback, $deferred, &$results, &$pending) {
-        if (0 === count($tasks)) {
+        if ($tasks instanceof \Iterator ? !$tasks->valid() : !$tasks) {
             $deferred->resolve($results);
             return;
         }
 
-        $task = array_shift($tasks);
-        $promise = call_user_func($task);
+        if ($tasks instanceof \Iterator) {
+            $task = $tasks->current();
+            $tasks->next();
+        } else {
+            $task = \array_shift($tasks);
+        }
+
+        $promise = \call_user_func($task);
         assert($promise instanceof PromiseInterface);
         $pending = $promise;
 
@@ -652,19 +656,26 @@ function waterfall(iterable $tasks): PromiseInterface
         $pending = null;
     });
 
-    if (!\is_array($tasks)) {
-        $tasks = \iterator_to_array($tasks);
+    if ($tasks instanceof \IteratorAggregate) {
+        $tasks = $tasks->getIterator();
+        assert($tasks instanceof \Iterator);
     }
 
     /** @var callable $next */
     $next = function ($value = null) use (&$tasks, &$next, $deferred, &$pending) {
-        if (0 === count($tasks)) {
+        if ($tasks instanceof \Iterator ? !$tasks->valid() : !$tasks) {
             $deferred->resolve($value);
             return;
         }
 
-        $task = array_shift($tasks);
-        $promise = call_user_func_array($task, func_get_args());
+        if ($tasks instanceof \Iterator) {
+            $task = $tasks->current();
+            $tasks->next();
+        } else {
+            $task = \array_shift($tasks);
+        }
+
+        $promise = \call_user_func_array($task, func_get_args());
         assert($promise instanceof PromiseInterface);
         $pending = $promise;
 

tests/ParallelTest.php

@@ -5,6 +5,7 @@
 use React;
 use React\EventLoop\Loop;
 use React\Promise\Promise;
+use function React\Promise\reject;
 
 class ParallelTest extends TestCase
 {
@@ -126,6 +127,25 @@ function () use (&$called) {
         $this->assertSame(2, $called);
     }
 
+    public function testParallelWithErrorFromInfiniteGeneratorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
+    {
+        $called = 0;
+
+        $tasks = (function () use (&$called) {
+            while (true) {
+                yield function () use (&$called) {
+                    return reject(new \RuntimeException('Rejected ' . ++$called));
+                };
+            }
+        })();
+
+        $promise = React\Async\parallel($tasks);
+
+        $promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));
+
+        $this->assertSame(1, $called);
+    }
+
     public function testParallelWithErrorWillCancelPendingPromises()
     {
         $cancelled = 0;

tests/SeriesTest.php

@@ -5,6 +5,7 @@
 use React;
 use React\EventLoop\Loop;
 use React\Promise\Promise;
+use function React\Promise\reject;
 
 class SeriesTest extends TestCase
 {
@@ -125,6 +126,47 @@ function () use (&$called) {
         $this->assertSame(1, $called);
     }
 
+    public function testSeriesWithErrorFromInfiniteGeneratorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
+    {
+        $called = 0;
+
+        $tasks = (function () use (&$called) {
+            while (true) {
+                yield function () use (&$called) {
+                    return reject(new \RuntimeException('Rejected ' . ++$called));
+                };
+            }
+        })();
+
+        $promise = React\Async\series($tasks);
+
+        $promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));
+
+        $this->assertSame(1, $called);
+    }
+
+    public function testSeriesWithErrorFromInfiniteIteratorAggregateReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
+    {
+        $tasks = new class() implements \IteratorAggregate {
+            public $called = 0;
+
+            public function getIterator(): \Iterator
+            {
+                while (true) {
+                    yield function () {
+                        return reject(new \RuntimeException('Rejected ' . ++$this->called));
+                    };
+                }
+            }
+        };
+
+        $promise = React\Async\series($tasks);
+
+        $promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));
+
+        $this->assertSame(1, $tasks->called);
+    }
+
     public function testSeriesWillCancelFirstPendingPromiseWhenCallingCancelOnResultingPromise()
     {
         $cancelled = 0;

tests/WaterfallTest.php

@@ -5,6 +5,7 @@
 use React;
 use React\EventLoop\Loop;
 use React\Promise\Promise;
+use function React\Promise\reject;
 
 class WaterfallTest extends TestCase
 {
@@ -139,6 +140,47 @@ function () use (&$called) {
         $this->assertSame(1, $called);
     }
 
+    public function testWaterfallWithErrorFromInfiniteGeneratorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
+    {
+        $called = 0;
+
+        $tasks = (function () use (&$called) {
+            while (true) {
+                yield function () use (&$called) {
+                    return reject(new \RuntimeException('Rejected ' . ++$called));
+                };
+            }
+        })();
+
+        $promise = React\Async\waterfall($tasks);
+
+        $promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));
+
+        $this->assertSame(1, $called);
+    }
+
+    public function testWaterfallWithErrorFromInfiniteIteratorAggregateReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
+    {
+        $tasks = new class() implements \IteratorAggregate {
+            public $called = 0;
+
+            public function getIterator(): \Iterator
+            {
+                while (true) {
+                    yield function () {
+                        return reject(new \RuntimeException('Rejected ' . ++$this->called));
+                    };
+                }
+            }
+        };
+
+        $promise = React\Async\waterfall($tasks);
+
+        $promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));
+
+        $this->assertSame(1, $tasks->called);
+    }
+
     public function testWaterfallWillCancelFirstPendingPromiseWhenCallingCancelOnResultingPromise()
     {
         $cancelled = 0;