Skip to content

[4.x] Support iterable type for parallel() + series() + waterfall() #49

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
WyriHaximus merged 2 commits into from
Jun 13, 2022
Merged

[4.x] Support iterable type for parallel() + series() + waterfall() #49

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a58b179
Support `iterable` type for `parallel()` + `series()` + `waterfall()`
clue Jun 5, 2022
2343d9c
Take advantage of iterators instead of converting to array first
clue Jun 11, 2022
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
Prev Previous commit
Take advantage of iterators instead of converting to array first
  • Loading branch information
@clue
clue committed Jun 12, 2022
commit 2343d9cd8cde75d2b253002ed1ddf5e84c714c5a
63 changes: 37 additions & 26 deletions src/functions.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -548,19 +548,10 @@ function parallel(iterable $tasks): PromiseInterface
$pending = [];
});
$results = [];
$errored = false;
$continue = true;

if (!\is_array($tasks)) {
$tasks = \iterator_to_array($tasks);
}

$numTasks = count($tasks);
if (0 === $numTasks) {
$deferred->resolve($results);
}

$taskErrback = function ($error) use (&$pending, $deferred, &$errored) {
$errored = true;
$taskErrback = function ($error) use (&$pending, $deferred, &$continue) {
$continue = false;
$deferred->reject($error);

foreach ($pending as $promise) {
Expand All @@ -572,25 +563,31 @@ function parallel(iterable $tasks): PromiseInterface
};

foreach ($tasks as $i => $task) {
$taskCallback = function ($result) use (&$results, &$pending, $numTasks, $i, $deferred) {
$taskCallback = function ($result) use (&$results, &$pending, &$continue, $i, $deferred) {
$results[$i] = $result;
unset($pending[$i]);

if (count($results) === $numTasks) {
if (!$pending && !$continue) {
$deferred->resolve($results);
}
};

$promise = call_user_func($task);
$promise = \call_user_func($task);
assert($promise instanceof PromiseInterface);
$pending[$i] = $promise;

$promise->then($taskCallback, $taskErrback);

if ($errored) {
if (!$continue) {
break;
}
}

$continue = false;
if (!$pending) {
$deferred->resolve($results);
}

return $deferred->promise();
}

Expand All @@ -609,8 +606,9 @@ function series(iterable $tasks): PromiseInterface
});
$results = [];

if (!\is_array($tasks)) {
$tasks = \iterator_to_array($tasks);
if ($tasks instanceof \IteratorAggregate) {
$tasks = $tasks->getIterator();
assert($tasks instanceof \Iterator);
}

/** @var callable():void $next */
Expand All @@ -620,13 +618,19 @@ function series(iterable $tasks): PromiseInterface
};

$next = function () use (&$tasks, $taskCallback, $deferred, &$results, &$pending) {
if (0 === count($tasks)) {
if ($tasks instanceof \Iterator ? !$tasks->valid() : !$tasks) {
$deferred->resolve($results);
return;
}

$task = array_shift($tasks);
$promise = call_user_func($task);
if ($tasks instanceof \Iterator) {
$task = $tasks->current();
$tasks->next();
} else {
$task = \array_shift($tasks);
}

$promise = \call_user_func($task);
assert($promise instanceof PromiseInterface);
$pending = $promise;

Expand All @@ -652,19 +656,26 @@ function waterfall(iterable $tasks): PromiseInterface
$pending = null;
});

if (!\is_array($tasks)) {
$tasks = \iterator_to_array($tasks);
if ($tasks instanceof \IteratorAggregate) {
$tasks = $tasks->getIterator();
assert($tasks instanceof \Iterator);
}

/** @var callable $next */
$next = function ($value = null) use (&$tasks, &$next, $deferred, &$pending) {
if (0 === count($tasks)) {
if ($tasks instanceof \Iterator ? !$tasks->valid() : !$tasks) {
$deferred->resolve($value);
return;
}

$task = array_shift($tasks);
$promise = call_user_func_array($task, func_get_args());
if ($tasks instanceof \Iterator) {
$task = $tasks->current();
$tasks->next();
} else {
$task = \array_shift($tasks);
}

$promise = \call_user_func_array($task, func_get_args());
assert($promise instanceof PromiseInterface);
$pending = $promise;

Expand Down
20 changes: 20 additions & 0 deletions tests/ParallelTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
use React;
use React\EventLoop\Loop;
use React\Promise\Promise;
use function React\Promise\reject;

class ParallelTest extends TestCase
{
Expand Down Expand Up @@ -126,6 +127,25 @@ function () use (&$called) {
$this->assertSame(2, $called);
}

public function testParallelWithErrorFromInfiniteGeneratorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
{
$called = 0;

$tasks = (function () use (&$called) {
while (true) {
yield function () use (&$called) {
return reject(new \RuntimeException('Rejected ' . ++$called));
};
}
})();

$promise = React\Async\parallel($tasks);

$promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));

$this->assertSame(1, $called);
}

public function testParallelWithErrorWillCancelPendingPromises()
{
$cancelled = 0;
Expand Down
42 changes: 42 additions & 0 deletions tests/SeriesTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
use React;
use React\EventLoop\Loop;
use React\Promise\Promise;
use function React\Promise\reject;

class SeriesTest extends TestCase
{
Expand Down Expand Up @@ -125,6 +126,47 @@ function () use (&$called) {
$this->assertSame(1, $called);
}

public function testSeriesWithErrorFromInfiniteGeneratorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
{
$called = 0;

$tasks = (function () use (&$called) {
while (true) {
yield function () use (&$called) {
return reject(new \RuntimeException('Rejected ' . ++$called));
};
}
})();

$promise = React\Async\series($tasks);

$promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));

$this->assertSame(1, $called);
}

public function testSeriesWithErrorFromInfiniteIteratorAggregateReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
{
$tasks = new class() implements \IteratorAggregate {
public $called = 0;

public function getIterator(): \Iterator
{
while (true) {
yield function () {
return reject(new \RuntimeException('Rejected ' . ++$this->called));
};
}
}
};

$promise = React\Async\series($tasks);

$promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));

$this->assertSame(1, $tasks->called);
}

public function testSeriesWillCancelFirstPendingPromiseWhenCallingCancelOnResultingPromise()
{
$cancelled = 0;
Expand Down
42 changes: 42 additions & 0 deletions tests/WaterfallTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
use React;
use React\EventLoop\Loop;
use React\Promise\Promise;
use function React\Promise\reject;

class WaterfallTest extends TestCase
{
Expand Down Expand Up @@ -139,6 +140,47 @@ function () use (&$called) {
$this->assertSame(1, $called);
}

public function testWaterfallWithErrorFromInfiniteGeneratorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
{
$called = 0;

$tasks = (function () use (&$called) {
while (true) {
yield function () use (&$called) {
return reject(new \RuntimeException('Rejected ' . ++$called));
};
}
})();

$promise = React\Async\waterfall($tasks);

$promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));

$this->assertSame(1, $called);
}

public function testWaterfallWithErrorFromInfiniteIteratorAggregateReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
{
$tasks = new class() implements \IteratorAggregate {
public $called = 0;

public function getIterator(): \Iterator
{
while (true) {
yield function () {
return reject(new \RuntimeException('Rejected ' . ++$this->called));
};
}
}
};

$promise = React\Async\waterfall($tasks);

$promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('Rejected 1')));

$this->assertSame(1, $tasks->called);
}

public function testWaterfallWillCancelFirstPendingPromiseWhenCallingCancelOnResultingPromise()
{
$cancelled = 0;
Expand Down