Skip to content

Commit ae86562

Browse files
authored
Merge pull request #43 from clue-labs/queue-closure
Support PHP 8.2, refactor queuing logic
2 parents 74c1d0c + a40d6ac commit ae86562

File tree

2 files changed

+34
-25
lines changed

2 files changed

+34
-25
lines changed

.github/workflows/ci.yml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ jobs:
1111
strategy:
1212
matrix:
1313
php:
14+
- 8.2
1415
- 8.1
1516
- 8.0
1617
- 7.4

src/Queue.php

Lines changed: 33 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,10 @@ class Queue implements \Countable
2727
private $limit;
2828
private $handler;
2929

30+
/** @var int<0,max> */
3031
private $pending = 0;
32+
33+
/** @var array<int,\Closure():void> */
3134
private $queue = array();
3235

3336
/**
@@ -373,24 +376,42 @@ public function __invoke()
373376
$id = key($queue);
374377
assert(is_int($id));
375378

376-
$deferred = new Deferred(function ($_, $reject) use (&$queue, $id, &$deferred) {
379+
/** @var ?PromiseInterface<T> $pending */
380+
$pending = null;
381+
382+
$deferred = new Deferred(function ($_, $reject) use (&$queue, $id, &$pending) {
377383
// forward cancellation to pending operation if it is currently executing
378-
if (isset($deferred->pending) && $deferred->pending instanceof PromiseInterface && \method_exists($deferred->pending, 'cancel')) {
379-
$deferred->pending->cancel();
384+
if ($pending instanceof PromiseInterface && \method_exists($pending, 'cancel')) {
385+
$pending->cancel();
380386
}
381-
unset($deferred->pending);
387+
$pending = null;
382388

383-
if (isset($deferred->args)) {
389+
if (isset($queue[$id])) {
384390
// queued promise cancelled before its handler is invoked
385391
// remove from queue and reject explicitly
386-
unset($queue[$id], $deferred->args);
392+
unset($queue[$id]);
387393
$reject(new \RuntimeException('Cancelled queued job before processing started'));
388394
}
389395
});
390396

391397
// queue job to process if number of pending jobs is below concurrency limit again
392-
$deferred->args = func_get_args();
393-
$queue[$id] = $deferred;
398+
$handler = $this->handler; // PHP 5.4+
399+
$args = func_get_args();
400+
$that = $this; // PHP 5.4+
401+
$queue[$id] = function () use ($handler, $args, $deferred, &$pending, $that) {
402+
$pending = \call_user_func_array($handler, $args);
403+
404+
$that->await($pending)->then(
405+
function ($result) use ($deferred, &$pending) {
406+
$pending = null;
407+
$deferred->resolve($result);
408+
},
409+
function ($e) use ($deferred, &$pending) {
410+
$pending = null;
411+
$deferred->reject($e);
412+
}
413+
);
414+
};
394415

395416
return $deferred->promise();
396417
}
@@ -407,7 +428,7 @@ public function count()
407428
*/
408429
public function await(PromiseInterface $promise)
409430
{
410-
$that = $this;
431+
$that = $this; // PHP 5.4+
411432

412433
return $promise->then(function ($result) use ($that) {
413434
$that->processQueue();
@@ -430,28 +451,15 @@ public function processQueue()
430451
return;
431452
}
432453

433-
$deferred = reset($this->queue);
434-
assert($deferred instanceof Deferred);
454+
$next = reset($this->queue);
455+
assert($next instanceof \Closure);
435456
unset($this->queue[key($this->queue)]);
436457

437458
// once number of pending jobs is below concurrency limit again:
438459
// await this situation, invoke handler and await its resolution before invoking next queued job
439460
++$this->pending;
440461

441-
$promise = call_user_func_array($this->handler, $deferred->args);
442-
$deferred->pending = $promise;
443-
unset($deferred->args);
444-
445462
// invoke handler and await its resolution before invoking next queued job
446-
$this->await($promise)->then(
447-
function ($result) use ($deferred) {
448-
unset($deferred->pending);
449-
$deferred->resolve($result);
450-
},
451-
function ($e) use ($deferred) {
452-
unset($deferred->pending);
453-
$deferred->reject($e);
454-
}
455-
);
463+
$next();
456464
}
457465
}

0 commit comments

Comments
 (0)