diff --git a/README.md b/README.md index 66f1b24c..00108b0f 100644 --- a/README.md +++ b/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( diff --git a/src/functions.php b/src/functions.php index eed32ddb..d6f2178d 100644 --- a/src/functions.php +++ b/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 $closures + * + * @return iterable> 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. * diff --git a/test/ConcurrentTest.php b/test/ConcurrentTest.php new file mode 100644 index 00000000..070a6df3 --- /dev/null +++ b/test/ConcurrentTest.php @@ -0,0 +1,90 @@ + 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'); + } +}