Add concurrent() and settle() functions

README.md

@@ -219,7 +219,20 @@ throws an exception.
 
 #### Combinators
 
-In concurrent applications, there will be multiple futures, where you might want to await them all or just the first one.
+In concurrent applications, there will be multiple futures, where you might want to await them all or just the first
+one.
+
+You can create a bunch of futures by applying `Amp\concurrent()` to an array of closures:
+it returns a `Future` for each, preserving the keys.
+
+```php
+$firstReachedApi = [
+    fn () => $httpClient->request(new Request('https://a.api.com', 'HEAD')),
+    fn () => $httpClient->request(new Request('https://b.api.com', 'HEAD')),
+] |> \Amp\concurrent(...) |> \Amp\Future\awaitAny(...);
+```
+
+The combinators below await such futures in different ways.
 
 ##### await
 
@@ -241,17 +254,16 @@ use Amp\Http\Client\Request;
 require __DIR__ . '/vendor/autoload.php';
 
 $httpClient = HttpClientBuilder::buildDefault();
-$uris = [
-    "google" => "https://www.google.com",
-    "news"   => "https://news.google.com",
-    "bing"   => "https://www.bing.com",
-    "yahoo"  => "https://www.yahoo.com",
-];
+
+$futures = Amp\concurrent([
+    "google" => fn () => $httpClient->request(new Request("https://www.google.com", 'HEAD')),
+    "news"   => fn () => $httpClient->request(new Request("https://news.google.com", 'HEAD')),
+    "bing"   => fn () => $httpClient->request(new Request("https://www.bing.com", 'HEAD')),
+    "yahoo"  => fn () => $httpClient->request(new Request("https://www.yahoo.com", 'HEAD')),
+]);
 
 try {
-    $responses = Future\await(array_map(function ($uri) use ($httpClient) {
-        return Amp\async(fn () => $httpClient->request(new Request($uri, 'HEAD')));
-    }, $uris));
+    $responses = Future\await($futures);
 
     foreach ($responses as $key => $response) {
         printf(
@@ -270,9 +282,38 @@ try {
 
 ##### awaitAnyN
 
-`Amp\Future\awaitAnyN($count, $iterable, $cancellation)` is the same as `await()` except that it tolerates individual errors. A result is returned once
-exactly `$count` instances in the `iterable` complete successfully. The return value is an array of values. The
-individual keys in the component array are preserved from the `iterable` passed to the function for evaluation.
+`Amp\Future\awaitAnyN($count, $iterable, $cancellation)` is the same as `await()` except that it tolerates individual errors.
+A result is returned once exactly `$count` instances in the `iterable` complete successfully, or `CompositeException` is thrown otherwise.
+The return value is an array of values with the individual keys preserved from the `iterable` passed to the function for evaluation.
+
+##### settle
+
+`Amp\Future\settle($iterable, $cancellation)` is the same as `await()` except that it waits for all the futures to finish (either successfully complete or error),
+not failing right off with the first error.
+
+A result is returned only if all the futures complete successfully.
+If at least one of them errors, a `CompositeException` comprised of all the errors is thrown.
+
+```php
+use Amp\CompositeException;
+use function Amp\concurrent;
+use function Amp\Future\settle;
+
+try {
+    $services = [
+        'db' => fn () => $pool->connect(),
+        'redis' => fn () => $cache->connect(),
+        'broker' => fn () => $amqp->connect(),
+        'secrets' => fn () => $vault->connect(),
+    ] |> concurrent(...) |> settle(...);
+} catch (CompositeException $e) {
+    $failedServices = array_keys($e->getReasons());
+    $message = sprintf('Services connection failed: %s.', implode(', ', $failedServices));
+
+    // Services connection failed: redis, broker.
+    throw new ServicesConnectionFailedException($message, $e->getReasons());
+}
+```
 
 ##### awaitAll
 

src/Future/functions.php

@@ -155,3 +155,32 @@ function await(iterable $futures, ?Cancellation $cancellation = null): array
     /** @var array<Tk, Tv> */
     return $values;
 }
+
+/**
+ * Waits for all futures to finish (either complete or error), unlike {@see await()}, which aborts on the first error.
+ *
+ * If any errored, a {@see CompositeException} holding all errors (keyed as given) is thrown.
+ * Otherwise, the unwrapped values are returned.
+ *
+ * The returned values are ordered by completion, not by the input iterable, just as with {@see await()}.
+ *
+ * @template Tk of array-key
+ * @template Tv
+ *
+ * @param iterable<Tk, Future<Tv>> $futures
+ * @param Cancellation|null $cancellation Optional cancellation.
+ *
+ * @return array<Tk, Tv> Unwrapped values, keyed as given.
+ *
+ * @throws CompositeException If one or more futures errored.
+ */
+function settle(iterable $futures, ?Cancellation $cancellation = null): array
+{
+    [$errors, $values] = awaitAll($futures, $cancellation);
+
+    if ($errors) {
+        throw new CompositeException($errors);
+    }
+
+    return $values;
+}

src/functions.php

@@ -43,6 +43,23 @@ function async(\Closure $closure, mixed ...$args): Future
     return new Future($state);
 }
 
+/**
+ * Concurrently evaluates the given closures, returning a {@see Future} for each.
+ *
+ * Pass or pipe the result into a combinator such as {@see Future\await()} or {@see Future\settle()} to await the values.
+ *
+ * @template Tk of array-key
+ * @template Tv
+ *
+ * @param array<Tk, \Closure():Tv> $closures
+ *
+ * @return array<Tk, Future<Tv>> A Future for each closure, with keys preserved.
+ */
+function concurrent(array $closures): array
+{
+    return \array_map(async(...), $closures);
+}
+
 /**
  * Returns the current time relative to an arbitrary point in time.
  *

test/ConcurrentTest.php

@@ -0,0 +1,50 @@
+<?php declare(strict_types=1);
+
+namespace Amp;
+
+class ConcurrentTest extends TestCase
+{
+    public function testEmpty(): void
+    {
+        self::assertSame([], concurrent([]));
+    }
+
+    public function testReturnsFuturePerClosureWithKeysPreserved(): void
+    {
+        $closures = ['one' => fn () => 1, 'two' => fn () => 2];
+
+        $futures = concurrent($closures);
+
+        self::assertContainsOnlyInstancesOf(Future::class, $futures);
+        self::assertSame(['one', 'two'], \array_keys($futures));
+    }
+
+    public function testClosuresAreEvaluated(): void
+    {
+        self::assertSame(
+            ['one' => 1, 'two' => 2],
+            Future\await(concurrent(['one' => fn () => 1, 'two' => fn () => 2]))
+        );
+    }
+
+    public function testClosuresAreEvaluatedConcurrently(): void
+    {
+        $order = [];
+
+        $futures = concurrent([
+            static function () use (&$order): void {
+                delay(0.02);
+                $order[] = 'slow';
+            },
+            static function () use (&$order): void {
+                delay(0.01);
+                $order[] = 'fast';
+            },
+        ]);
+
+        Future\await($futures);
+
+        // "fast" is declared second but completes first, proving the closures run concurrently.
+        self::assertSame(['fast', 'slow'], $order);
+    }
+}

test/Future/SettleTest.php

@@ -0,0 +1,126 @@
+<?php declare(strict_types=1);
+
+namespace Amp\Future;
+
+use Amp\CancelledException;
+use Amp\CompositeException;
+use Amp\DeferredFuture;
+use Amp\Future;
+use Amp\TimeoutCancellation;
+use PHPUnit\Framework\TestCase;
+use Revolt\EventLoop;
+
+class SettleTest extends TestCase
+{
+    public function testEmpty(): void
+    {
+        self::assertSame([], settle([]));
+    }
+
+    public function testAllComplete(): void
+    {
+        $futures = [
+            Future::complete(1),
+            Future::complete(2),
+        ];
+
+        self::assertSame([1, 2], settle($futures));
+    }
+
+    public function testKeysPreserved(): void
+    {
+        $futures = [
+            'one' => Future::complete(1),
+            'two' => Future::complete(2),
+        ];
+
+        self::assertSame(
+            ['one' => 1, 'two' => 2],
+            settle($futures),
+        );
+    }
+
+    public function testSingleError(): void
+    {
+        $exception = new \Exception('foo');
+        $futures = [
+            'one' => Future::error($exception),
+            'two' => Future::complete(2),
+        ];
+
+        try {
+            settle($futures);
+
+            self::fail('Expected ' . CompositeException::class . ' to be thrown');
+        } catch (CompositeException $composite) {
+            self::assertSame(
+                ['one' => $exception],
+                $composite->getReasons(),
+            );
+        }
+    }
+
+    public function testMultipleErrors(): void
+    {
+        $first = new \Exception('foo');
+        $second = new \RuntimeException('bar');
+        $futures = [
+            Future::error($first),
+            Future::error($second),
+        ];
+
+        try {
+            settle($futures);
+
+            self::fail('Expected ' . CompositeException::class . ' to be thrown');
+        } catch (CompositeException $composite) {
+            self::assertSame(
+                [$first, $second],
+                $composite->getReasons(),
+            );
+        }
+    }
+
+    public function testWaitsForAllBeforeThrowing(): void
+    {
+        $immediate = new \Exception('foo');
+        $delayed = new \RuntimeException('bar');
+
+        $deferred = new DeferredFuture;
+        EventLoop::delay(0.01, fn () => $deferred->error($delayed));
+
+        $futures = [
+            'delayed' => $deferred->getFuture(),
+            'immediate' => Future::error($immediate),
+        ];
+
+        try {
+            settle($futures);
+
+            self::fail('Expected ' . CompositeException::class . ' to be thrown');
+        } catch (CompositeException $composite) {
+            $reasons = $composite->getReasons();
+
+            self::assertSame([
+                'immediate' => $immediate,
+                'delayed' => $delayed,
+            ], $reasons);
+        }
+    }
+
+    public function testCancellation(): void
+    {
+        $this->expectException(CancelledException::class);
+
+        $deferreds = \array_map(function (int $value) {
+            $deferred = new DeferredFuture;
+            EventLoop::delay($value / 10, fn () => $deferred->complete($value));
+            return $deferred;
+        }, \range(1, 3));
+
+        settle(\array_map(
+            fn (DeferredFuture $deferred) => $deferred->getFuture(),
+            $deferreds
+        ), new TimeoutCancellation(0.05));
+    }
+}