From 85eb94823bf576ee3ed761280022abf5bc02728b Mon Sep 17 00:00:00 2001 From: Yevhen Sidelnyk Date: Thu, 26 Mar 2026 17:01:24 +0200 Subject: [PATCH] Add Amp\disperse() function --- README.md | 11 +++++ src/functions.php | 32 ++++++++++++++ test/Future/DisperseTest.php | 81 ++++++++++++++++++++++++++++++++++++ 3 files changed, 124 insertions(+) create mode 100644 test/Future/DisperseTest.php diff --git a/README.md b/README.md index 66f1b24c..579758a6 100644 --- a/README.md +++ b/README.md @@ -278,6 +278,17 @@ individual keys in the component array are preserved from the `iterable` passed `Amp\Future\awaitAll($iterable, $cancellation)` awaits all futures and returns their results as `[$errors, $values]` array. +##### disperse + +`Amp\disperse($closures, $cancellation)` executes all given closures asynchronously and returns their results on completion. + +```php +$results = \Amp\disperse([ + fn () => $httpClient->request(new Request('https://www.google.com', 'HEAD')), + fn () => $httpClient->request(new Request('https://www.bing.com', 'HEAD')), +]); +``` + ##### awaitFirst `Amp\Future\awaitFirst($iterable, $cancellation)` unwraps the first completed `Future`, whether successfully completed or errored. diff --git a/src/functions.php b/src/functions.php index eed32ddb..2411d64f 100644 --- a/src/functions.php +++ b/src/functions.php @@ -5,6 +5,8 @@ use Revolt\EventLoop; use Revolt\EventLoop\UnsupportedFeatureException; +use function Amp\Future\awaitAll; + /** * Creates a new fiber to execute the given closure asynchronously. A Future is returned which is completed with the * return value of the passed closure or will fail if the closure throws an exception. @@ -43,6 +45,36 @@ function async(\Closure $closure, mixed ...$args): Future return new Future($state); } +/** + * Executes the given closures concurrently and returns their results as soon as all closures complete successfully. + * + * @template Tk of array-key + * @template Tv + * + * @param array $closures + * @param Cancellation|null $cancellation Optional cancellation. + * + * @return array Results in input order of Closures. + */ +function disperse(array $closures, ?Cancellation $cancellation = null): array +{ + if ([] === $closures) { + return []; + } + + $futures = \array_map(static fn (\Closure $closure): Future => async($closure), $closures); + + [$errors, $values] = awaitAll($futures, $cancellation); + + if ($errors) { + /** @var non-empty-array $errors */ + throw new CompositeException($errors); + } + + /** @var non-empty-array */ + return \array_replace($closures, $values); +} + /** * Returns the current time relative to an arbitrary point in time. * diff --git a/test/Future/DisperseTest.php b/test/Future/DisperseTest.php new file mode 100644 index 00000000..2d484b79 --- /dev/null +++ b/test/Future/DisperseTest.php @@ -0,0 +1,81 @@ + 1, + fn () => 2, + ])); + } + + public function testCompletionOrder(): void + { + $result = disperse([ + 'slow' => function (): string { + delay(0.05); + return 'slow'; + }, + 'fast' => function (): string { + delay(0.01); + return 'fast'; + }, + ]); + + self::assertSame([ + 'slow' => 'slow', + 'fast' => 'fast', + ], $result); + } + + public function testNonClosure(): void + { + $this->expectException(\TypeError::class); + + disperse(['not-a-closure']); + } + + public function testErrors(): void + { + try { + disperse([ + fn () => 1, + fn () => throw new \Exception('boom'), + ]); + + $this->fail('The code should have thrown'); + } catch (CompositeException $e) { + } + + $reasons = $e->getReasons(); + self::assertCount(1, $reasons); + + self::assertSame('boom', $reasons[1]->getMessage()); + } + + public function testCancellation(): void + { + $this->expectException(CancelledException::class); + + disperse([ + function (): int { + delay(0.05); + return 1; + }, + function (): int { + delay(0.05); + return 2; + }, + ], new TimeoutCancellation(0.01)); + } + + public function testEmpty(): void + { + $this->assertSame([], disperse([])); + } +}