Ensure parallel() function stops invoking additional tasks on error

Author: clue

src/functions.php

@@ -102,7 +102,7 @@ function parallel(array $tasks)
 
     $done = function () use (&$results, &$errors, $deferred) {
         if (count($errors)) {
-            $deferred->reject(array_shift($errors));
+            $deferred->reject(reset($errors));
             return;
         }
 
@@ -116,7 +116,7 @@ function parallel(array $tasks)
     }
 
     $checkDone = function () use (&$results, &$errors, $numTasks, $done) {
-        if ($numTasks === count($results) + count($errors)) {
+        if ($numTasks === count($results) || count($errors)) {
             $done();
         }
     };
@@ -136,6 +136,10 @@ function parallel(array $tasks)
         assert($promise instanceof PromiseInterface);
 
         $promise->then($taskCallback, $taskErrback);
+
+        if ($errors) {
+            break;
+        }
     }
 
     return $deferred->promise();

tests/ParallelTest.php

@@ -29,7 +29,7 @@ function () {
             },
             function () {
                 return new Promise(function ($resolve) {
-                    Loop::addTimer(0.1, function () use ($resolve) {
+                    Loop::addTimer(0.11, function () use ($resolve) {
                         $resolve('bar');
                     });
                 });
@@ -49,7 +49,7 @@ function () {
         $timer->assertInRange(0.1, 0.2);
     }
 
-    public function testParallelWithError()
+    public function testParallelWithErrorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
     {
         $called = 0;
 
@@ -60,7 +60,8 @@ function () use (&$called) {
                     $resolve('foo');
                 });
             },
-            function () {
+            function () use (&$called) {
+                $called++;
                 return new Promise(function () {
                     throw new \RuntimeException('whoops');
                 });
@@ -80,7 +81,7 @@ function () use (&$called) {
         $this->assertSame(2, $called);
     }
 
-    public function testParallelWithDelayedError()
+    public function testParallelWithDelayedErrorReturnsPromiseRejectedWithExceptionFromTask()
     {
         $called = 0;
 
@@ -91,7 +92,8 @@ function () use (&$called) {
                     $resolve('foo');
                 });
             },
-            function () {
+            function () use (&$called) {
+                $called++;
                 return new Promise(function ($_, $reject) {
                     Loop::addTimer(0.001, function () use ($reject) {
                         $reject(new \RuntimeException('whoops'));
@@ -112,6 +114,6 @@ function () use (&$called) {
 
         Loop::run();
 
-        $this->assertSame(2, $called);
+        $this->assertSame(3, $called);
     }
 }