Add concurrent() function

README.md

@@ -219,7 +219,22 @@ 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 iterable of closures:
+it returns a `Future` for each, preserving the keys.
+
+```php
+$firstReachableApiResponse = \Amp\Future\awaitAny([
+    fn () => $httpClient->request(new Request('https://a.api.com', 'HEAD')),
+    fn () => $httpClient->request(new Request('https://b.api.com', 'HEAD')),
+] |> \Amp\concurrent(...));
+```
+
+`Amp\concurrent()` returns `\Generator`, so producers of unbounded length are supported.
+
+The combinators below await such futures in different ways.
 
 ##### await
 
@@ -241,17 +256,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(

src/functions.php

@@ -43,6 +43,26 @@ function async(\Closure $closure, mixed ...$args): Future
     return new Future($state);
 }
 
+/**
+ * Concurrently evaluates the given closures, returning a {@see Future} for each.
+ * Each closure is started as the iterable is consumed, so producers of unknown (even unbounded) length are supported.
+ *
+ * Pass or pipe the result into a combinator such as {@see Future\await()} to await the values.
+ *
+ * @template Tk
+ * @template Tv
+ *
+ * @param iterable<Tk, \Closure():Tv> $closures
+ *
+ * @return iterable<Tk, Future<Tv>> A Future for each closure, with keys preserved.
+ */
+function concurrent(iterable $closures): iterable
+{
+    foreach ($closures as $key => $closure) {
+        yield $key => async($closure);
+    }
+}
+
 /**
  * Returns the current time relative to an arbitrary point in time.
  *

test/ConcurrentTest.php

@@ -0,0 +1,90 @@
+<?php declare(strict_types=1);
+
+namespace Amp;
+
+class ConcurrentTest extends TestCase
+{
+    public function testEmpty(): void
+    {
+        self::assertSame([], \iterator_to_array(concurrent([])));
+    }
+
+    public function testReturnsFuturePerClosureWithKeysPreserved(): void
+    {
+        $closures = ['one' => fn () => 1, 'two' => fn () => 2];
+
+        $futures = \iterator_to_array(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);
+    }
+
+    public function testGeneratorInput(): void
+    {
+        $closures = (static function (): \Generator {
+            yield 'one' => static fn () => 1;
+            yield 'two' => static fn () => 2;
+        })();
+
+        self::assertSame(['one' => 1, 'two' => 2], Future\await(concurrent($closures)));
+    }
+
+    public function testUnboundedProducer(): void
+    {
+        $closures = (static function (): \Generator {
+            for ($i = 0; ; ++$i) {
+                delay(0.01);
+
+                yield static fn (): int => $i;
+            }
+        })();
+
+        self::assertSame([0, 1], Future\awaitAnyN(2, concurrent($closures)));
+    }
+
+    public function testCombinatorConsumesTheEntireConcurrentIterator(): void
+    {
+        $consumed = 0;
+
+        $closures = (static function () use (&$consumed): \Generator {
+            while (++$consumed < 3) {
+                yield static fn () => 7;
+            }
+        })();
+
+        $futures = concurrent($closures);
+        self::assertSame(0, $consumed, 'Nothing is consumed till combinator is called');
+
+        self::assertSame(7, Future\awaitFirst($futures));
+        self::assertSame(3, $consumed, 'Combinator consumes all the iterations');
+    }
+}