Support iterable type for parallel() + series() + waterfall()

Author: clue

README.md

@@ -396,7 +396,7 @@ $promise->then(function (int $bytes) {
 
 ### parallel()
 
-The `parallel(array<callable():PromiseInterface<mixed,Exception>> $tasks): PromiseInterface<array<mixed>,Exception>` function can be used
+The `parallel(iterable<callable():PromiseInterface<mixed,Exception>> $tasks): PromiseInterface<array<mixed>,Exception>` function can be used
 like this:
 
 ```php
@@ -438,7 +438,7 @@ React\Async\parallel([
 
 ### series()
 
-The `series(array<callable():PromiseInterface<mixed,Exception>> $tasks): PromiseInterface<array<mixed>,Exception>` function can be used
+The `series(iterable<callable():PromiseInterface<mixed,Exception>> $tasks): PromiseInterface<array<mixed>,Exception>` function can be used
 like this:
 
 ```php
@@ -480,7 +480,7 @@ React\Async\series([
 
 ### waterfall()
 
-The `waterfall(array<callable(mixed=):PromiseInterface<mixed,Exception>> $tasks): PromiseInterface<mixed,Exception>` function can be used
+The `waterfall(iterable<callable(mixed=):PromiseInterface<mixed,Exception>> $tasks): PromiseInterface<mixed,Exception>` function can be used
 like this:
 
 ```php

src/functions.php

@@ -533,10 +533,10 @@ function coroutine(callable $function, mixed ...$args): PromiseInterface
 }
 
 /**
- * @param array<callable():PromiseInterface<mixed,Exception>> $tasks
+ * @param iterable<callable():PromiseInterface<mixed,Exception>> $tasks
  * @return PromiseInterface<array<mixed>,Exception>
  */
-function parallel(array $tasks): PromiseInterface
+function parallel(iterable $tasks): PromiseInterface
 {
     $pending = [];
     $deferred = new Deferred(function () use (&$pending) {
@@ -550,6 +550,10 @@ function parallel(array $tasks): PromiseInterface
     $results = [];
     $errored = false;
 
+    if (!\is_array($tasks)) {
+        $tasks = \iterator_to_array($tasks);
+    }
+
     $numTasks = count($tasks);
     if (0 === $numTasks) {
         $deferred->resolve($results);
@@ -591,10 +595,10 @@ function parallel(array $tasks): PromiseInterface
 }
 
 /**
- * @param array<callable():PromiseInterface<mixed,Exception>> $tasks
+ * @param iterable<callable():PromiseInterface<mixed,Exception>> $tasks
  * @return PromiseInterface<array<mixed>,Exception>
  */
-function series(array $tasks): PromiseInterface
+function series(iterable $tasks): PromiseInterface
 {
     $pending = null;
     $deferred = new Deferred(function () use (&$pending) {
@@ -605,6 +609,10 @@ function series(array $tasks): PromiseInterface
     });
     $results = [];
 
+    if (!\is_array($tasks)) {
+        $tasks = \iterator_to_array($tasks);
+    }
+
     /** @var callable():void $next */
     $taskCallback = function ($result) use (&$results, &$next) {
         $results[] = $result;
@@ -631,10 +639,10 @@ function series(array $tasks): PromiseInterface
 }
 
 /**
- * @param array<callable(mixed=):PromiseInterface<mixed,Exception>> $tasks
+ * @param iterable<callable(mixed=):PromiseInterface<mixed,Exception>> $tasks
  * @return PromiseInterface<mixed,Exception>
  */
-function waterfall(array $tasks): PromiseInterface
+function waterfall(iterable $tasks): PromiseInterface
 {
     $pending = null;
     $deferred = new Deferred(function () use (&$pending) {
@@ -644,6 +652,10 @@ function waterfall(array $tasks): PromiseInterface
         $pending = null;
     });
 
+    if (!\is_array($tasks)) {
+        $tasks = \iterator_to_array($tasks);
+    }
+
     /** @var callable $next */
     $next = function ($value = null) use (&$tasks, &$next, $deferred, &$pending) {
         if (0 === count($tasks)) {

tests/ParallelTest.php

@@ -17,6 +17,19 @@ public function testParallelWithoutTasks()
         $promise->then($this->expectCallableOnceWith(array()));
     }
 
+    public function testParallelWithoutTasksFromEmptyGeneratorResolvesWithEmptyArray()
+    {
+        $tasks = (function () {
+            if (false) {
+                yield;
+            }
+        })();
+
+        $promise = React\Async\parallel($tasks);
+
+        $promise->then($this->expectCallableOnceWith([]));
+    }
+
     public function testParallelWithTasks()
     {
         $tasks = array(
@@ -49,6 +62,38 @@ function () {
         $timer->assertInRange(0.1, 0.2);
     }
 
+    public function testParallelWithTasksFromGeneratorResolvesWithArrayOfFulfillmentValues()
+    {
+        $tasks = (function () {
+            yield function () {
+                return new Promise(function ($resolve) {
+                    Loop::addTimer(0.1, function () use ($resolve) {
+                        $resolve('foo');
+                    });
+                });
+            };
+            yield function () {
+                return new Promise(function ($resolve) {
+                    Loop::addTimer(0.11, function () use ($resolve) {
+                        $resolve('bar');
+                    });
+                });
+            };
+        })();
+
+        $promise = React\Async\parallel($tasks);
+
+        $promise->then($this->expectCallableOnceWith(array('foo', 'bar')));
+
+        $timer = new Timer($this);
+        $timer->start();
+
+        Loop::run();
+
+        $timer->stop();
+        $timer->assertInRange(0.1, 0.2);
+    }
+
     public function testParallelWithErrorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
     {
         $called = 0;

tests/SeriesTest.php

@@ -17,6 +17,19 @@ public function testSeriesWithoutTasks()
         $promise->then($this->expectCallableOnceWith(array()));
     }
 
+    public function testSeriesWithoutTasksFromEmptyGeneratorResolvesWithEmptyArray()
+    {
+        $tasks = (function () {
+            if (false) {
+                yield;
+            }
+        })();
+
+        $promise = React\Async\series($tasks);
+
+        $promise->then($this->expectCallableOnceWith([]));
+    }
+
     public function testSeriesWithTasks()
     {
         $tasks = array(
@@ -49,6 +62,38 @@ function () {
         $timer->assertInRange(0.10, 0.20);
     }
 
+    public function testSeriesWithTasksFromGeneratorResolvesWithArrayOfFulfillmentValues()
+    {
+        $tasks = (function () {
+            yield function () {
+                return new Promise(function ($resolve) {
+                    Loop::addTimer(0.051, function () use ($resolve) {
+                        $resolve('foo');
+                    });
+                });
+            };
+            yield function () {
+                return new Promise(function ($resolve) {
+                    Loop::addTimer(0.051, function () use ($resolve) {
+                        $resolve('bar');
+                    });
+                });
+            };
+        })();
+
+        $promise = React\Async\series($tasks);
+
+        $promise->then($this->expectCallableOnceWith(array('foo', 'bar')));
+
+        $timer = new Timer($this);
+        $timer->start();
+
+        Loop::run();
+
+        $timer->stop();
+        $timer->assertInRange(0.10, 0.20);
+    }
+
     public function testSeriesWithError()
     {
         $called = 0;

tests/WaterfallTest.php

@@ -17,6 +17,19 @@ public function testWaterfallWithoutTasks()
         $promise->then($this->expectCallableOnceWith(null));
     }
 
+    public function testWaterfallWithoutTasksFromEmptyGeneratorResolvesWithNull()
+    {
+        $tasks = (function () {
+            if (false) {
+                yield;
+            }
+        })();
+
+        $promise = React\Async\waterfall($tasks);
+
+        $promise->then($this->expectCallableOnceWith(null));
+    }
+
     public function testWaterfallWithTasks()
     {
         $tasks = array(
@@ -56,6 +69,45 @@ function ($bar) {
         $timer->assertInRange(0.15, 0.30);
     }
 
+    public function testWaterfallWithTasksFromGeneratorResolvesWithFinalFulfillmentValue()
+    {
+        $tasks = (function () {
+            yield function ($foo = 'foo') {
+                return new Promise(function ($resolve) use ($foo) {
+                    Loop::addTimer(0.05, function () use ($resolve, $foo) {
+                        $resolve($foo);
+                    });
+                });
+            };
+            yield function ($foo) {
+                return new Promise(function ($resolve) use ($foo) {
+                    Loop::addTimer(0.05, function () use ($resolve, $foo) {
+                        $resolve($foo . 'bar');
+                    });
+                });
+            };
+            yield function ($bar) {
+                return new Promise(function ($resolve) use ($bar) {
+                    Loop::addTimer(0.05, function () use ($resolve, $bar) {
+                        $resolve($bar . 'baz');
+                    });
+                });
+            };
+        })();
+
+        $promise = React\Async\waterfall($tasks);
+
+        $promise->then($this->expectCallableOnceWith('foobarbaz'));
+
+        $timer = new Timer($this);
+        $timer->start();
+
+        Loop::run();
+
+        $timer->stop();
+        $timer->assertInRange(0.15, 0.30);
+    }
+
     public function testWaterfallWithError()
     {
         $called = 0;