Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
95.68%
133 / 139
84.62%
22 / 26
 CRAP
0.00%
0 / 1
Queue
95.68%
133 / 139
84.62%
22 / 26
 67
0.00%
0 / 1
 __construct
100.00%
4 / 4
100.00%
1 / 1
 2
 create
100.00%
1 / 1
100.00%
1 / 1
 1
 fake
100.00%
1 / 1
100.00%
1 / 1
 1
 setPriority
100.00%
2 / 2
100.00%
1 / 1
 1
 getPriority
100.00%
1 / 1
100.00%
1 / 1
 1
 isFifo
100.00%
1 / 1
100.00%
1 / 1
 1
 isFilo
100.00%
1 / 1
100.00%
1 / 1
 1
 isLilo
100.00%
1 / 1
100.00%
1 / 1
 1
 isLifo
100.00%
1 / 1
100.00%
1 / 1
 1
 setEvents
100.00%
2 / 2
100.00%
1 / 1
 1
 getEvents
100.00%
1 / 1
100.00%
1 / 1
 1
 events
100.00%
1 / 1
100.00%
1 / 1
 1
 hasEvents
100.00%
1 / 1
100.00%
1 / 1
 1
 triggerEvent
100.00%
4 / 4
100.00%
1 / 1
 4
 addJob
100.00%
4 / 4
100.00%
1 / 1
 2
 addJobs
100.00%
3 / 3
100.00%
1 / 1
 2
 addTask
75.00%
6 / 8
0.00%
0 / 1
 4.25
 addTasks
100.00%
3 / 3
100.00%
1 / 1
 2
 work
100.00%
27 / 27
100.00%
1 / 1
 7
 runWithTimeout
100.00%
12 / 12
100.00%
1 / 1
 4
 evaluateTasksOnce
93.94%
31 / 33
0.00%
0 / 1
 14.04
 getScheduledTasks
66.67%
2 / 3
0.00%
0 / 1
 2.15
 run
94.12%
16 / 17
0.00%
0 / 1
 8.01
 clear
100.00%
2 / 2
100.00%
1 / 1
 1
 clearFailed
100.00%
2 / 2
100.00%
1 / 1
 1
 clearTasks
100.00%
3 / 3
100.00%
1 / 1
 2
1<?php
2declare(strict_types=1);
3/**
4 * Pop PHP Framework (https://www.popphp.org/)
5 *
6 * @link       https://github.com/popphp/popphp-framework
7 * @author     Nick Sagona, III <nick@popphp.org>
8 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
9 * @license    https://www.popphp.org/license     New BSD License
10 */
11
12/**
13 * @namespace
14 */
15namespace Pop\Queue;
16
17use Pop\Application;
18use Pop\Event\Manager as EventManager;
19use Pop\Queue\Adapter\AdapterInterface;
20use Pop\Queue\Adapter\Memory;
21use Pop\Queue\Adapter\TaskAdapterInterface;
22use Pop\Queue\Process\AbstractJob;
23use Pop\Queue\Process\Task;
24use Pop\Queue\Process\TimeoutException;
25
26/**
27 * Queue class
28 *
29 * @category   Pop
30 * @package    Pop\Queue
31 * @author     Nick Sagona, III <nick@popphp.org>
32 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
33 * @license    https://www.popphp.org/license     New BSD License
34 * @version    3.0.0
35 */
36class Queue extends AbstractQueue
37{
38
39    /**
40     * Queue priority constants
41     */
42    const FIFO = 'FIFO'; // Same as LILO
43    const FILO = 'FILO'; // Same as LIFO
44
45    /**
46     * Event manager, for lifecycle observability hooks (queue.job.*, queue.task.*).
47     * If not set, and an Application with its own event manager is passed into
48     * work()/run(), that Application's event manager is used instead - see
49     * triggerEvent(). If neither is available, event firing is a silent no-op.
50     * @var ?EventManager
51     */
52    protected ?EventManager $events = null;
53
54    /**
55     * Constructor
56     *
57     * Instantiate the queue object
58     *
59     * @param string $name
60     * @param AdapterInterface|TaskAdapterInterface $adapter
61     * @param ?string $priority
62     */
63    public function __construct(string $name, AdapterInterface|TaskAdapterInterface $adapter, ?string $priority = null)
64    {
65        $this->setName($name);
66        $this->setAdapter($adapter);
67        if ($priority !== null) {
68            $this->setPriority($priority);
69        }
70    }
71
72    /**
73     * Create the queue object
74     *
75     * @param  string $name
76     * @param  AdapterInterface|TaskAdapterInterface $adapter
77     * @param  ?string $priority
78     * @return Queue
79     */
80    public static function create(
81        string $name, AdapterInterface|TaskAdapterInterface $adapter, ?string $priority = null
82    ): Queue
83    {
84        return new self($name, $adapter, $priority);
85    }
86
87    /**
88     * Create a Memory-backed queue for testing - a fake, in the sense
89     * familiar from other PHP frameworks' testing conventions. Note
90     * Memory's own constructor takes $leaseSeconds before $priority
91     * (Memory predates this method and that argument order is documented,
92     * pre-existing behavior elsewhere in this codebase) - fake()'s own
93     * parameter order matches create()'s $priority-before-lease convention
94     * instead, and translates between the two internally, so a caller of
95     * fake() never needs to know Memory's own argument order.
96     *
97     * @param  string  $name
98     * @param  ?string $priority
99     * @param  int     $leaseSeconds
100     * @return Queue
101     */
102    public static function fake(string $name = 'pop-queue', ?string $priority = null, int $leaseSeconds = 60): Queue
103    {
104        return new self($name, new Memory($leaseSeconds, $priority));
105    }
106
107    /**
108     * Set queue priority
109     *
110     * @param  string $priority
111     * @return Queue
112     */
113    public function setPriority(string $priority = 'FIFO'): Queue
114    {
115        $this->adapter->setPriority($priority);
116        return $this;
117    }
118
119    /**
120     * Get queue priority
121     *
122     * @return string
123     */
124    public function getPriority(): string
125    {
126        return $this->adapter->getPriority();
127    }
128
129    /**
130     * Is FIFO
131     *
132     * @return bool
133     */
134    public function isFifo(): bool
135    {
136        return $this->adapter->isFifo();
137    }
138
139    /**
140     * Is FILO
141     *
142     * @return bool
143     */
144    public function isFilo(): bool
145    {
146        return $this->adapter->isFilo();
147    }
148
149    /**
150     * Is LILO (alias to FIFO)
151     *
152     * @return bool
153     */
154    public function isLilo(): bool
155    {
156        return $this->adapter->isLilo();
157    }
158
159    /**
160     * Is LIFO (alias to FILO)
161     *
162     * @return bool
163     */
164    public function isLifo(): bool
165    {
166        return $this->adapter->isLifo();
167    }
168
169    /**
170     * Set event manager
171     *
172     * @param  EventManager $events
173     * @return Queue
174     */
175    public function setEvents(EventManager $events): Queue
176    {
177        $this->events = $events;
178        return $this;
179    }
180
181    /**
182     * Get event manager
183     *
184     * @return ?EventManager
185     */
186    public function getEvents(): ?EventManager
187    {
188        return $this->events;
189    }
190
191    /**
192     * Get event manager (alias)
193     *
194     * @return ?EventManager
195     */
196    public function events(): ?EventManager
197    {
198        return $this->events;
199    }
200
201    /**
202     * Has event manager
203     *
204     * @return bool
205     */
206    public function hasEvents(): bool
207    {
208        return ($this->events !== null);
209    }
210
211    /**
212     * Trigger a lifecycle event. Uses this Queue's own event manager if one
213     * is set via setEvents(); otherwise falls back to $application's event
214     * manager if one was passed in and has events registered; otherwise
215     * does nothing. Never throws on its own account - if the resolved
216     * manager's trigger() call throws (e.g. a listener's own code throws),
217     * that exception propagates to the caller exactly as any other
218     * uncaught exception would.
219     *
220     * @param  string       $name
221     * @param  array        $params
222     * @param  ?Application $application
223     * @return void
224     */
225    protected function triggerEvent(string $name, array $params, ?Application $application = null): void
226    {
227        if ($this->hasEvents()) {
228            $this->events->trigger($name, $params);
229        } else if (($application !== null) && ($application->events() !== null)) {
230            $application->events()->trigger($name, $params);
231        }
232    }
233
234    /**
235     * Add job
236     *
237     * @param  AbstractJob $job
238     * @param  ?int        $maxAttempts
239     * @return Queue
240     */
241    public function addJob(AbstractJob $job, ?int $maxAttempts = null): Queue
242    {
243        if ($maxAttempts !== null) {
244            $job->setMaxAttempts($maxAttempts);
245        }
246        $this->adapter->push($job);
247
248        return $this;
249    }
250
251    /**
252     * Add jobs
253     *
254     * @param  array $jobs
255     * @param  ?int  $maxAttempts
256     * @return Queue
257     */
258    public function addJobs(array $jobs, ?int $maxAttempts = null): Queue
259    {
260        foreach ($jobs as $job) {
261            $this->addJob($job, $maxAttempts);
262        }
263        return $this;
264    }
265
266    /**
267     * Add task (alias)
268     *
269     * @param  Task $task
270     * @param  ?int $maxAttempts
271     * @param  ?int $gracePeriod
272     * @throws Exception
273     * @return Queue
274     */
275    public function addTask(Task $task, ?int $maxAttempts = null, ?int $gracePeriod = null): Queue
276    {
277        if (!($this->adapter instanceof TaskAdapterInterface)) {
278            throw new Exception('Error: That queue adapter does not support scheduled tasks');
279        }
280        if ($maxAttempts !== null) {
281            $task->setMaxAttempts($maxAttempts);
282        }
283        if ($gracePeriod !== null) {
284            $task->setGracePeriod($gracePeriod);
285        }
286
287        $this->adapter->schedule($task);
288
289        return $this;
290    }
291
292    /**
293     * Add tasks
294     *
295     * @param  array $tasks
296     * @param  ?int  $maxAttempts
297     * @throws Exception
298     * @return Queue
299     */
300    public function addTasks(array $tasks, ?int $maxAttempts = null): Queue
301    {
302        foreach ($tasks as $task) {
303            $this->addTask($task, $maxAttempts);
304        }
305        return $this;
306    }
307
308    /**
309     * Work next job
310     *
311     * @param  ?Application $application
312     * @return ?AbstractJob
313     */
314    public function work(?Application $application = null): ?AbstractJob
315    {
316        $job = $this->adapter->reserve();
317        if ($job === null) {
318            return null;
319        }
320
321        if (!$job->isValid()) {
322            $reason = 'Exceeded max attempts or expired before execution';
323            $this->adapter->bury($job, $reason);
324            $this->triggerEvent('queue.job.buried', ['job' => $job, 'queue' => $this, 'reason' => $reason], $application);
325            return $job;
326        }
327
328        $this->triggerEvent('queue.job.pre', ['job' => $job, 'queue' => $this], $application);
329
330        $exception = null;
331        $buried    = false;
332
333        try {
334            $this->runWithTimeout($job, $application);
335            $job->complete();
336            $this->adapter->delete($job);
337        } catch (\Throwable $e) {
338            $exception = $e;
339            $job->failed($e->getMessage());
340            if ($job->isValid()) {
341                $this->adapter->release($job);
342            } else {
343                $buried = true;
344                $this->adapter->bury($job, $e->getMessage());
345            }
346        }
347
348        if ($exception === null) {
349            $this->triggerEvent('queue.job.post', ['job' => $job, 'queue' => $this], $application);
350        } else {
351            $this->triggerEvent('queue.job.failed', ['job' => $job, 'queue' => $this, 'exception' => $exception], $application);
352            if ($buried) {
353                $this->triggerEvent('queue.job.buried', ['job' => $job, 'queue' => $this, 'reason' => $exception->getMessage()], $application);
354            }
355        }
356
357        return $job;
358    }
359
360    /**
361     * Run a job, enforcing its soft timeout when ext-pcntl is available.
362     * Exec-type jobs are exempt from this pcntl-based alarm entirely -
363     * AbstractJob::runExec() wires the job's timeout directly into
364     * Symfony\Process's own setTimeout(), which is both more precise and
365     * (unlike a pcntl alarm racing against a raw exec() call) actually
366     * capable of killing the spawned child process on expiry. Layering a
367     * second, competing pcntl alarm on top for the same job risks
368     * interrupting Process's own internal wait()/kill logic mid-flight if
369     * the alarm fires first, defeating the reliable-child-termination
370     * guarantee this task exists to add.
371     *
372     * @param  AbstractJob  $job
373     * @param  ?Application $application
374     * @return mixed
375     */
376    protected function runWithTimeout(AbstractJob $job, ?Application $application): mixed
377    {
378        if (!$job->hasTimeout() || !extension_loaded('pcntl') || $job->hasExec()) {
379            return $job->run($application);
380        }
381
382        pcntl_async_signals(true);
383        pcntl_signal(SIGALRM, function() use ($job) {
384            throw new TimeoutException(
385                'Error: Job ' . $job->getJobId() . ' exceeded its ' . $job->getTimeout() . ' second timeout.'
386            );
387        });
388        pcntl_alarm($job->getTimeout());
389
390        try {
391            return $job->run($application);
392        } finally {
393            pcntl_alarm(0);
394            pcntl_signal(SIGALRM, SIG_DFL);
395        }
396    }
397
398    /**
399     * Evaluate every given task exactly once against the current time and
400     * run the ones that are due, returning the ones that ran (successfully
401     * or not) keyed by job ID. Coarse (non-sub-minute) tasks are always
402     * considered; pass $onlySubMinute = true to skip them (used by run()'s
403     * per-second tick loop, where a coarse task's single evaluation already
404     * happened on the shared first pass and doesn't need repeating).
405     *
406     * Before running a due task, atomically claims it for the current
407     * due-window via the adapter (claimTaskRun()) - if another worker
408     * sharing the same adapter storage already claimed this task's window,
409     * this call skips it silently rather than running it a second time.
410     *
411     * @param  array        $tasks         taskId => Task
412     * @param  ?Application $application
413     * @param  bool         $onlySubMinute
414     * @return array  jobId => Task, for every task that ran this pass
415     */
416    public function evaluateTasksOnce(array $tasks, ?Application $application, bool $onlySubMinute = false): array
417    {
418        if (!($this->adapter instanceof TaskAdapterInterface)) {
419            return [];
420        }
421
422        $ran = [];
423
424        foreach ($tasks as $taskId => $task) {
425            $isSubMinute = $task->cron()->hasSeconds();
426
427            if ($onlySubMinute && !$isSubMinute) {
428                continue;
429            }
430
431            if ($isSubMinute) {
432                $task->__wakeup();
433            }
434
435            if ((!$task->isValid()) || (!$task->cron()->evaluate())) {
436                continue;
437            }
438
439            $window = $isSubMinute ? (string)time() : (string)intdiv(time(), 60);
440            if (!$this->adapter->claimTaskRun($taskId, $window)) {
441                // Another worker already claimed this task's current
442                // due-window - not a failure, just not ours to run.
443                continue;
444            }
445
446            $this->triggerEvent('queue.task.pre', ['task' => $task, 'queue' => $this], $application);
447
448            $exception = null;
449
450            try {
451                $task->run($application);
452                $task->complete();
453                if (!$isSubMinute) {
454                    $this->adapter->updateTask($task);
455                }
456            } catch (\Exception $e) {
457                $exception = $e;
458                $task->failed($e->getMessage());
459                if ($isSubMinute) {
460                    $this->adapter->removeTask($taskId);
461                    $this->adapter->schedule($task);
462                    $this->adapter->claimTaskRun($taskId, $window);
463                } else {
464                    $this->adapter->updateTask($task);
465                }
466            }
467
468            $ran[$task->getJobId()] = $task;
469
470            if ($exception === null) {
471                $this->triggerEvent('queue.task.post', ['task' => $task, 'queue' => $this], $application);
472            } else {
473                $this->triggerEvent('queue.task.failed', ['task' => $task, 'queue' => $this, 'exception' => $exception], $application);
474            }
475        }
476
477        return $ran;
478    }
479
480    /**
481     * Fetch every currently scheduled task from the adapter, once. Returns
482     * an empty array if the adapter doesn't support tasks
483     * (TaskAdapterInterface) or has none scheduled - callers don't need to
484     * duplicate that guard.
485     *
486     * @return array  taskId => Task
487     */
488    public function getScheduledTasks(): array
489    {
490        if (!($this->adapter instanceof TaskAdapterInterface)) {
491            return [];
492        }
493
494        // One call rather than a listing followed by a fetch per task. That
495        // pattern cost a round trip per scheduled task, and this runs on every
496        // run() - so once a minute per worker per queue, forever. The hasTasks()
497        // pre-check went with it: it was a second round trip asking a question
498        // this call already answers by coming back empty.
499        return $this->adapter->getAllTasks();
500    }
501
502    /**
503     * Run schedule
504     *
505     * Evaluates every scheduled task fairly: all tasks get one shared
506     * evaluation pass immediately, then - only if at least one sub-minute
507     * task exists - up to 59 more passes (one per second) considering only
508     * the sub-minute tasks. This ensures no single task's per-tick work
509     * blocks any other task's evaluation on the same tick.
510     *
511     * @param  ?Application $application
512     * @throws Process\Exception
513     * @return array
514     */
515    public function run(?Application $application = null): array
516    {
517        $tasks = [];
518
519        $scheduledTasks = $this->getScheduledTasks();
520        if (empty($scheduledTasks)) {
521            return $tasks;
522        }
523
524        foreach ($this->evaluateTasksOnce($scheduledTasks, $application) as $jobId => $task) {
525            $tasks[$jobId] = $task;
526        }
527
528        $hasSubMinute = false;
529        foreach ($scheduledTasks as $task) {
530            if ($task->cron()->hasSeconds()) {
531                $hasSubMinute = true;
532                break;
533            }
534        }
535
536        if ($hasSubMinute) {
537            for ($tick = 1; $tick < 60; $tick++) {
538                sleep(1);
539                foreach ($this->evaluateTasksOnce($scheduledTasks, $application, true) as $jobId => $task) {
540                    $tasks[$jobId] = $task;
541                }
542            }
543        }
544
545        return $tasks;
546    }
547
548    /**
549     * Clear jobs from queue
550     *
551     * @return Queue
552     */
553    public function clear(): Queue
554    {
555        $this->adapter->clear();
556        return $this;
557    }
558
559    /**
560     * Clear dead-letter jobs from queue
561     *
562     * @return Queue
563     */
564    public function clearFailed(): Queue
565    {
566        $this->adapter->clearDead();
567        return $this;
568    }
569
570    /**
571     * Clear tasks from queue
572     *
573     * @return Queue
574     */
575    public function clearTasks(): Queue
576    {
577        if ($this->adapter instanceof TaskAdapterInterface) {
578            $this->adapter->clearTasks();
579        }
580        return $this;
581    }
582
583}