Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
32 changes: 32 additions & 0 deletions src/functions.php
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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<Tk, \Closure():Tv> $closures
* @param Cancellation|null $cancellation Optional cancellation.
*
* @return array<Tk, Tv> 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<array-key, \Throwable> $errors */
throw new CompositeException($errors);
}

/** @var non-empty-array<Tk, Tv> */
return \array_replace($closures, $values);
}

/**
* Returns the current time relative to an arbitrary point in time.
*
Expand Down
81 changes: 81 additions & 0 deletions test/Future/DisperseTest.php
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
<?php declare(strict_types=1);

namespace Amp;

use PHPUnit\Framework\TestCase;

class DisperseTest extends TestCase
{
public function testTwoComplete(): void
{
self::assertSame([1, 2], disperse([
fn () => 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([]));
}
}
Loading