Skip to content

Commit a112d86

Browse files
WyriHaximusWyriHaximus
authored
Merge pull request #9 from clue-labs/cancellation
2 parents 98ae760 + 36eb448 commit a112d86

4 files changed

Lines changed: 155 additions & 33 deletions

File tree

‎src/functions.php‎

Lines changed: 49 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace React\Async;
44

55
use React\EventLoop\Loop;
6+
use React\Promise\CancellablePromiseInterface;
67
use React\Promise\Deferred;
78
use React\Promise\PromiseInterface;
89

@@ -96,46 +97,53 @@ function ($error) use (&$exception, &$rejected, &$wait) {
9697
*/
9798
function parallel(array $tasks)
9899
{
99-
$deferred = new Deferred();
100-
$results = array();
101-
$errors = array();
102-
103-
$done = function () use (&$results, &$errors, $deferred) {
104-
if (count($errors)) {
105-
$deferred->reject(array_shift($errors));
106-
return;
100+
$pending = array();
101+
$deferred = new Deferred(function () use (&$pending) {
102+
foreach ($pending as $promise) {
103+
if ($promise instanceof CancellablePromiseInterface) {
104+
$promise->cancel();
105+
}
107106
}
108-
109-
$deferred->resolve($results);
110-
};
107+
$pending = array();
108+
});
109+
$results = array();
110+
$errored = false;
111111

112112
$numTasks = count($tasks);
113-
114113
if (0 === $numTasks) {
115-
$done();
114+
$deferred->resolve($results);
116115
}
117116

118-
$checkDone = function () use (&$results, &$errors, $numTasks, $done) {
119-
if ($numTasks === count($results) + count($errors)) {
120-
$done();
121-
}
122-
};
117+
$taskErrback = function ($error) use (&$pending, $deferred, &$errored) {
118+
$errored = true;
119+
$deferred->reject($error);
123120

124-
$taskErrback = function ($error) use (&$errors, $checkDone) {
125-
$errors[] = $error;
126-
$checkDone();
121+
foreach ($pending as $promise) {
122+
if ($promise instanceof CancellablePromiseInterface) {
123+
$promise->cancel();
124+
}
125+
}
126+
$pending = array();
127127
};
128128

129129
foreach ($tasks as $i => $task) {
130-
$taskCallback = function ($result) use (&$results, $i, $checkDone) {
130+
$taskCallback = function ($result) use (&$results, &$pending, $numTasks, $i, $deferred) {
131131
$results[$i] = $result;
132-
$checkDone();
132+
133+
if (count($results) === $numTasks) {
134+
$deferred->resolve($results);
135+
}
133136
};
134137

135138
$promise = call_user_func($task);
136139
assert($promise instanceof PromiseInterface);
140+
$pending[$i] = $promise;
137141

138142
$promise->then($taskCallback, $taskErrback);
143+
144+
if ($errored) {
145+
break;
146+
}
139147
}
140148

141149
return $deferred->promise();
@@ -147,7 +155,13 @@ function parallel(array $tasks)
147155
*/
148156
function series(array $tasks)
149157
{
150-
$deferred = new Deferred();
158+
$pending = null;
159+
$deferred = new Deferred(function () use (&$pending) {
160+
if ($pending instanceof CancellablePromiseInterface) {
161+
$pending->cancel();
162+
}
163+
$pending = null;
164+
});
151165
$results = array();
152166

153167
/** @var callable():void $next */
@@ -156,7 +170,7 @@ function series(array $tasks)
156170
$next();
157171
};
158172

159-
$next = function () use (&$tasks, $taskCallback, $deferred, &$results) {
173+
$next = function () use (&$tasks, $taskCallback, $deferred, &$results, &$pending) {
160174
if (0 === count($tasks)) {
161175
$deferred->resolve($results);
162176
return;
@@ -165,6 +179,7 @@ function series(array $tasks)
165179
$task = array_shift($tasks);
166180
$promise = call_user_func($task);
167181
assert($promise instanceof PromiseInterface);
182+
$pending = $promise;
168183

169184
$promise->then($taskCallback, array($deferred, 'reject'));
170185
};
@@ -180,10 +195,16 @@ function series(array $tasks)
180195
*/
181196
function waterfall(array $tasks)
182197
{
183-
$deferred = new Deferred();
198+
$pending = null;
199+
$deferred = new Deferred(function () use (&$pending) {
200+
if ($pending instanceof CancellablePromiseInterface) {
201+
$pending->cancel();
202+
}
203+
$pending = null;
204+
});
184205

185206
/** @var callable $next */
186-
$next = function ($value = null) use (&$tasks, &$next, $deferred) {
207+
$next = function ($value = null) use (&$tasks, &$next, $deferred, &$pending) {
187208
if (0 === count($tasks)) {
188209
$deferred->resolve($value);
189210
return;
@@ -192,6 +213,7 @@ function waterfall(array $tasks)
192213
$task = array_shift($tasks);
193214
$promise = call_user_func_array($task, func_get_args());
194215
assert($promise instanceof PromiseInterface);
216+
$pending = $promise;
195217

196218
$promise->then($next, array($deferred, 'reject'));
197219
};

‎tests/ParallelTest.php‎

Lines changed: 60 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ function () {
2929
},
3030
function () {
3131
return new Promise(function ($resolve) {
32-
Loop::addTimer(0.1, function () use ($resolve) {
32+
Loop::addTimer(0.11, function () use ($resolve) {
3333
$resolve('bar');
3434
});
3535
});
@@ -49,7 +49,7 @@ function () {
4949
$timer->assertInRange(0.1, 0.2);
5050
}
5151

52-
public function testParallelWithError()
52+
public function testParallelWithErrorReturnsPromiseRejectedWithExceptionFromTaskAndStopsCallingAdditionalTasks()
5353
{
5454
$called = 0;
5555

@@ -60,7 +60,8 @@ function () use (&$called) {
6060
$resolve('foo');
6161
});
6262
},
63-
function () {
63+
function () use (&$called) {
64+
$called++;
6465
return new Promise(function () {
6566
throw new \RuntimeException('whoops');
6667
});
@@ -80,7 +81,59 @@ function () use (&$called) {
8081
$this->assertSame(2, $called);
8182
}
8283

83-
public function testParallelWithDelayedError()
84+
public function testParallelWithErrorWillCancelPendingPromises()
85+
{
86+
$cancelled = 0;
87+
88+
$tasks = array(
89+
function () use (&$cancelled) {
90+
return new Promise(function () { }, function () use (&$cancelled) {
91+
$cancelled++;
92+
});
93+
},
94+
function () {
95+
return new Promise(function () {
96+
throw new \RuntimeException('whoops');
97+
});
98+
},
99+
function () use (&$cancelled) {
100+
return new Promise(function () { }, function () use (&$cancelled) {
101+
$cancelled++;
102+
});
103+
}
104+
);
105+
106+
$promise = React\Async\parallel($tasks);
107+
108+
$promise->then(null, $this->expectCallableOnceWith(new \RuntimeException('whoops')));
109+
110+
$this->assertSame(1, $cancelled);
111+
}
112+
113+
public function testParallelWillCancelPendingPromisesWhenCallingCancelOnResultingPromise()
114+
{
115+
$cancelled = 0;
116+
117+
$tasks = array(
118+
function () use (&$cancelled) {
119+
return new Promise(function () { }, function () use (&$cancelled) {
120+
$cancelled++;
121+
});
122+
},
123+
function () use (&$cancelled) {
124+
return new Promise(function () { }, function () use (&$cancelled) {
125+
$cancelled++;
126+
});
127+
}
128+
);
129+
130+
$promise = React\Async\parallel($tasks);
131+
$promise->cancel();
132+
133+
$this->assertSame(2, $cancelled);
134+
}
135+
136+
public function testParallelWithDelayedErrorReturnsPromiseRejectedWithExceptionFromTask()
84137
{
85138
$called = 0;
86139

@@ -91,7 +144,8 @@ function () use (&$called) {
91144
$resolve('foo');
92145
});
93146
},
94-
function () {
147+
function () use (&$called) {
148+
$called++;
95149
return new Promise(function ($_, $reject) {
96150
Loop::addTimer(0.001, function () use ($reject) {
97151
$reject(new \RuntimeException('whoops'));
@@ -112,6 +166,6 @@ function () use (&$called) {
112166

113167
Loop::run();
114168

115-
$this->assertSame(2, $called);
169+
$this->assertSame(3, $called);
116170
}
117171
}

‎tests/SeriesTest.php‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,4 +79,27 @@ function () use (&$called) {
7979

8080
$this->assertSame(1, $called);
8181
}
82+
83+
public function testSeriesWillCancelFirstPendingPromiseWhenCallingCancelOnResultingPromise()
84+
{
85+
$cancelled = 0;
86+
87+
$tasks = array(
88+
function () {
89+
return new Promise(function ($resolve) {
90+
$resolve();
91+
});
92+
},
93+
function () use (&$cancelled) {
94+
return new Promise(function () { }, function () use (&$cancelled) {
95+
$cancelled++;
96+
});
97+
}
98+
);
99+
100+
$promise = React\Async\series($tasks);
101+
$promise->cancel();
102+
103+
$this->assertSame(1, $cancelled);
104+
}
82105
}

‎tests/WaterfallTest.php‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,4 +86,27 @@ function () use (&$called) {
8686

8787
$this->assertSame(1, $called);
8888
}
89+
90+
public function testWaterfallWillCancelFirstPendingPromiseWhenCallingCancelOnResultingPromise()
91+
{
92+
$cancelled = 0;
93+
94+
$tasks = array(
95+
function () {
96+
return new Promise(function ($resolve) {
97+
$resolve();
98+
});
99+
},
100+
function () use (&$cancelled) {
101+
return new Promise(function () { }, function () use (&$cancelled) {
102+
$cancelled++;
103+
});
104+
}
105+
);
106+
107+
$promise = React\Async\waterfall($tasks);
108+
$promise->cancel();
109+
110+
$this->assertSame(1, $cancelled);
111+
}
89112
}

0 commit comments

Comments
 (0)