Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
98.92%
92 / 93
96.15%
25 / 26
 CRAP
0.00%
0 / 1
Memory
98.92%
92 / 93
96.15%
25 / 26
 43
0.00%
0 / 1
 __construct
100.00%
2 / 2
100.00%
1 / 1
 1
 push
100.00%
9 / 9
100.00%
1 / 1
 1
 reserve
100.00%
16 / 16
100.00%
1 / 1
 6
 release
88.89%
8 / 9
0.00%
0 / 1
 2.01
 delete
100.00%
3 / 3
100.00%
1 / 1
 1
 bury
100.00%
8 / 8
100.00%
1 / 1
 1
 hasJobs
100.00%
1 / 1
100.00%
1 / 1
 1
 count
100.00%
1 / 1
100.00%
1 / 1
 1
 clear
100.00%
3 / 3
100.00%
1 / 1
 1
 hasDeadJobs
100.00%
1 / 1
100.00%
1 / 1
 1
 countDead
100.00%
1 / 1
100.00%
1 / 1
 1
 getDeadJobs
100.00%
4 / 4
100.00%
1 / 1
 3
 getDeadJob
100.00%
3 / 3
100.00%
1 / 1
 3
 retryDeadJob
100.00%
4 / 4
100.00%
1 / 1
 2
 deleteDeadJob
100.00%
2 / 2
100.00%
1 / 1
 1
 clearDead
100.00%
2 / 2
100.00%
1 / 1
 1
 schedule
100.00%
3 / 3
100.00%
1 / 1
 2
 getTasks
100.00%
1 / 1
100.00%
1 / 1
 1
 getTask
100.00%
1 / 1
100.00%
1 / 1
 2
 getAllTasks
100.00%
1 / 1
100.00%
1 / 1
 1
 updateTask
100.00%
4 / 4
100.00%
1 / 1
 2
 removeTask
100.00%
2 / 2
100.00%
1 / 1
 1
 getTaskCount
100.00%
1 / 1
100.00%
1 / 1
 1
 hasTasks
100.00%
1 / 1
100.00%
1 / 1
 1
 clearTasks
100.00%
3 / 3
100.00%
1 / 1
 1
 claimTaskRun
100.00%
7 / 7
100.00%
1 / 1
 4
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\Adapter;
16
17use Pop\Queue\Process\AbstractJob;
18use Pop\Queue\Process\Task;
19
20/**
21 * In-memory adapter class
22 *
23 * A hermetic, single-process reference implementation of the adapter
24 * contract. Used to prove the contract in tests independent of any real
25 * storage backend, and as a fake for consumers of this package to test
26 * against.
27 *
28 * @category   Pop
29 * @package    Pop\Queue
30 * @author     Nick Sagona, III <nick@popphp.org>
31 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
32 * @license    https://www.popphp.org/license     New BSD License
33 * @version    3.0.0
34 */
35class Memory extends AbstractTaskAdapter
36{
37
38    /**
39     * Jobs, keyed by job ID
40     * @var array
41     */
42    protected array $jobs = [];
43
44    /**
45     * Job metadata (sequence, status, availableAt, reservedUntil), keyed by job ID
46     * @var array
47     */
48    protected array $meta = [];
49
50    /**
51     * Dead-letter jobs, keyed by job ID
52     * @var array
53     */
54    protected array $dead = [];
55
56    /**
57     * Scheduled tasks, keyed by task ID
58     * @var array
59     */
60    protected array $tasks = [];
61
62    /**
63     * Task claim state: taskId => [window, expiresAt]
64     * @var array
65     */
66    protected array $taskClaims = [];
67
68    /**
69     * Push/reserve ordering sequence counter
70     * @var int
71     */
72    protected int $sequence = 0;
73
74    /**
75     * Reservation lease length, in seconds
76     * @var int
77     */
78    protected int $leaseSeconds = 60;
79
80    /**
81     * Constructor
82     *
83     * @param int     $leaseSeconds
84     * @param ?string $priority
85     */
86    public function __construct(int $leaseSeconds = 60, ?string $priority = null)
87    {
88        $this->leaseSeconds = $leaseSeconds;
89        parent::__construct($priority);
90    }
91
92    public function push(AbstractJob $job): Memory
93    {
94        $jobId = $job->getJobId();
95        $this->jobs[$jobId] = clone $job;
96        $this->meta[$jobId] = [
97            'sequence'      => $this->sequence++,
98            'status'        => 'pending',
99            'availableAt'   => $job->getAvailableAt() ?? time(),
100            'reservedUntil' => null,
101        ];
102
103        return $this;
104    }
105
106    public function reserve(): ?AbstractJob
107    {
108        $now      = time();
109        $eligible = array_filter($this->meta, function($meta) use ($now) {
110            return ($meta['availableAt'] <= $now) && (
111                ($meta['status'] === 'pending') ||
112                (($meta['status'] === 'reserved') && ($meta['reservedUntil'] <= $now))
113            );
114        });
115
116        if (empty($eligible)) {
117            return null;
118        }
119
120        uasort($eligible, function($a, $b) {
121            return $this->isFilo() ? ($b['sequence'] <=> $a['sequence']) : ($a['sequence'] <=> $b['sequence']);
122        });
123
124        $jobId = array_key_first($eligible);
125
126        $this->meta[$jobId]['status']        = 'reserved';
127        $this->meta[$jobId]['reservedUntil'] = $now + $this->leaseSeconds;
128
129        return clone $this->jobs[$jobId];
130    }
131
132    public function release(AbstractJob $job, ?int $delay = null): Memory
133    {
134        $jobId = $job->getJobId();
135        if (!isset($this->meta[$jobId])) {
136            return $this;
137        }
138
139        $delay = $delay ?? $job->getBackoffDelay();
140
141        $this->jobs[$jobId] = clone $job;
142        $this->meta[$jobId]['status']        = 'pending';
143        $this->meta[$jobId]['availableAt']   = time() + $delay;
144        $this->meta[$jobId]['reservedUntil'] = null;
145
146        return $this;
147    }
148
149    public function delete(AbstractJob $job): Memory
150    {
151        $jobId = $job->getJobId();
152        unset($this->jobs[$jobId], $this->meta[$jobId]);
153
154        return $this;
155    }
156
157    public function bury(AbstractJob $job, ?string $reason = null): Memory
158    {
159        $jobId = $job->getJobId();
160        unset($this->jobs[$jobId], $this->meta[$jobId]);
161
162        $this->dead[$jobId] = [
163            'job'      => clone $job,
164            'reason'   => $reason,
165            'buriedAt' => time(),
166        ];
167
168        return $this;
169    }
170
171    public function hasJobs(): bool
172    {
173        return !empty($this->jobs);
174    }
175
176    public function count(): int
177    {
178        return count($this->jobs);
179    }
180
181    public function clear(): Memory
182    {
183        $this->jobs = [];
184        $this->meta = [];
185
186        return $this;
187    }
188
189    public function hasDeadJobs(): bool
190    {
191        return !empty($this->dead);
192    }
193
194    public function countDead(): int
195    {
196        return count($this->dead);
197    }
198
199    public function getDeadJobs(bool $unserialize = true): array
200    {
201        $jobs = [];
202        foreach ($this->dead as $jobId => $entry) {
203            $jobs[$jobId] = $unserialize ? $entry['job'] : $entry;
204        }
205
206        return $jobs;
207    }
208
209    public function getDeadJob(string $jobId, bool $unserialize = true): mixed
210    {
211        if (!isset($this->dead[$jobId])) {
212            return null;
213        }
214
215        return $unserialize ? $this->dead[$jobId]['job'] : $this->dead[$jobId];
216    }
217
218    public function retryDeadJob(string $jobId): Memory
219    {
220        if (isset($this->dead[$jobId])) {
221            $this->push($this->dead[$jobId]['job']);
222            unset($this->dead[$jobId]);
223        }
224
225        return $this;
226    }
227
228    public function deleteDeadJob(string $jobId): Memory
229    {
230        unset($this->dead[$jobId]);
231
232        return $this;
233    }
234
235    public function clearDead(): Memory
236    {
237        $this->dead = [];
238
239        return $this;
240    }
241
242    public function schedule(Task $task): Memory
243    {
244        if ($task->isValid()) {
245            $this->tasks[$task->getJobId()] = clone $task;
246        }
247
248        return $this;
249    }
250
251    public function getTasks(): array
252    {
253        return array_keys($this->tasks);
254    }
255
256    public function getTask(string $taskId): ?Task
257    {
258        return isset($this->tasks[$taskId]) ? clone $this->tasks[$taskId] : null;
259    }
260
261    public function getAllTasks(): array
262    {
263        // Cloned per task for the same reason getTask() clones: callers must not
264        // get a handle on the stored object and mutate the queue by accident.
265        return array_map(fn(Task $task) => clone $task, $this->tasks);
266    }
267
268    public function updateTask(Task $task): Memory
269    {
270        if ($task->isValid()) {
271            $this->tasks[$task->getJobId()] = clone $task;
272        } else {
273            $this->removeTask($task->getJobId());
274        }
275
276        return $this;
277    }
278
279    public function removeTask(string $taskId): Memory
280    {
281        unset($this->tasks[$taskId], $this->taskClaims[$taskId]);
282
283        return $this;
284    }
285
286    public function getTaskCount(): int
287    {
288        return count($this->tasks);
289    }
290
291    public function hasTasks(): bool
292    {
293        return !empty($this->tasks);
294    }
295
296    public function clearTasks(): Memory
297    {
298        $this->tasks       = [];
299        $this->taskClaims  = [];
300
301        return $this;
302    }
303
304    public function claimTaskRun(string $taskId, string $window): bool
305    {
306        $now = time();
307
308        if (isset($this->taskClaims[$taskId])) {
309            [$claimedWindow, $expiresAt] = $this->taskClaims[$taskId];
310            if (($claimedWindow === $window) && ($expiresAt > $now)) {
311                return false;
312            }
313        }
314
315        $this->taskClaims[$taskId] = [$window, $now + self::TASK_CLAIM_TTL];
316
317        return true;
318    }
319
320}