Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
89.36%
42 / 47
81.82%
9 / 11
 CRAP
0.00%
0 / 1
Redis
89.36%
42 / 47
81.82%
9 / 11
 26.81
0.00%
0 / 1
 __construct
62.50%
5 / 8
0.00%
0 / 1
 6.32
 getRedis
100.00%
1 / 1
100.00%
1 / 1
 1
 getPrefix
100.00%
1 / 1
100.00%
1 / 1
 1
 recordKey
100.00%
1 / 1
100.00%
1 / 1
 1
 workerSetKey
100.00%
1 / 1
100.00%
1 / 1
 1
 ensureWorkerSet
100.00%
10 / 10
100.00%
1 / 1
 5
 write
100.00%
2 / 2
100.00%
1 / 1
 1
 read
100.00%
2 / 2
100.00%
1 / 1
 2
 all
81.82%
9 / 11
0.00%
0 / 1
 4.10
 delete
100.00%
2 / 2
100.00%
1 / 1
 1
 purgeUndecodable
100.00%
8 / 8
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\Registry\Adapter;
16
17use Pop\Queue\Registry\AbstractRegistry;
18use Pop\Queue\Registry\Exception;
19use Pop\Queue\Registry\WorkerRecord;
20
21/**
22 * Redis registry class
23 *
24 * One string key per worker, "{prefix}:worker:{id}", holding the JSON
25 * record.
26 *
27 * @category   Pop
28 * @package    Pop\Queue
29 * @author     Nick Sagona, III <nick@popphp.org>
30 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
31 * @license    https://www.popphp.org/license     New BSD License
32 * @version    3.0.0
33 */
34class Redis extends AbstractRegistry
35{
36
37    /**
38     * Redis object
39     * @var \Redis|null
40     */
41    protected \Redis|null $redis = null;
42
43    /**
44     * Key prefix
45     * @var string
46     */
47    protected string $prefix = 'pop-registry';
48
49    /**
50     * Whether this instance has already reconciled the worker index set against
51     * any record keys written without it. See ensureWorkerSet().
52     * @var bool
53     */
54    protected bool $workerSetChecked = false;
55
56    /**
57     * Constructor
58     *
59     * @param  string     $host
60     * @param  int|string $port
61     * @param  string     $prefix
62     * @param  ?string    $password
63     * @param  ?array     $context
64     * @throws Exception|\RedisException
65     */
66    public function __construct(
67        string $host = 'localhost', int|string $port = 6379, string $prefix = 'pop-registry',
68        ?string $password = null, ?array $context = null
69    )
70    {
71        if (!class_exists('Redis', false)) {
72            throw new Exception('Error: Redis is not available.');
73        }
74
75        $this->redis  = new \Redis();
76        $this->prefix = $prefix;
77
78        if (!$this->redis->connect($host, (int)$port, context: $context)) {
79            throw new Exception('Error: Unable to connect to the redis server.');
80        }
81
82        if (($password !== null) && !$this->redis->auth($password)) {
83            throw new Exception('Error: Unable to authenticate with the redis server.');
84        }
85    }
86
87    /**
88     * Get the Redis object
89     *
90     * @return \Redis|null
91     */
92    public function getRedis(): \Redis|null
93    {
94        return $this->redis;
95    }
96
97    /**
98     * Get the key prefix
99     *
100     * @return string
101     */
102    public function getPrefix(): string
103    {
104        return $this->prefix;
105    }
106
107    /**
108     * The key backing a given worker ID
109     *
110     * @param  string $id
111     * @return string
112     */
113    protected function recordKey(string $id): string
114    {
115        return $this->prefix . ':worker:' . $id;
116    }
117
118    /**
119     * Key of the set indexing registered worker IDs.
120     *
121     * Enumerating the registry used to mean a KEYS scan, which Redis runs
122     * against its whole keyspace while blocking every other client. all() is
123     * called by the stuck-worker sweep, so that scan landed on a live server on
124     * a routine schedule. Membership is tracked in a set instead, making the
125     * enumeration one SMEMBERS against a single key.
126     *
127     * @return string
128     */
129    protected function workerSetKey(): string
130    {
131        return $this->prefix . ':workers';
132    }
133
134    /**
135     * Bring the worker index set in line with any record keys that aren't in it,
136     * once per adapter instance.
137     *
138     * A record key can exist outside the set two ways: it was written by a
139     * version of this adapter that predates the set, or something wrote the key
140     * directly. Either way the set is now the only thing all() and
141     * purgeUndecodable() consult, so an unindexed record would be invisible to
142     * the stuck-worker sweep and to pruning - it would sit there forever,
143     * unreadable and unreapable.
144     *
145     * The reconciliation costs one KEYS scan per Redis database, guarded by a
146     * marker key, which is the one place this adapter still issues one.
147     *
148     * @return void
149     */
150    protected function ensureWorkerSet(): void
151    {
152        if ($this->workerSetChecked) {
153            return;
154        }
155        $this->workerSetChecked = true;
156
157        if ($this->redis->exists($this->prefix . ':index-built')) {
158            return;
159        }
160
161        foreach ($this->redis->keys($this->prefix . ':worker:*') as $key) {
162            $id = substr($key, (strrpos($key, ':worker:') + 8));
163            if ($id !== '') {
164                $this->redis->sAdd($this->workerSetKey(), $id);
165            }
166        }
167
168        $this->redis->set($this->prefix . ':index-built', '1');
169    }
170
171    public function write(WorkerRecord $record): void
172    {
173        $this->redis->set($this->recordKey($record->getId()), $this->encode($record));
174        $this->redis->sAdd($this->workerSetKey(), $record->getId());
175    }
176
177    public function read(string $id): ?WorkerRecord
178    {
179        $value = $this->redis->get($this->recordKey($id));
180
181        return ($value !== false) ? $this->decode($value) : null;
182    }
183
184    public function all(): array
185    {
186        $this->ensureWorkerSet();
187
188        $records = [];
189
190        foreach ($this->redis->sMembers($this->workerSetKey()) as $id) {
191            $value = $this->redis->get($this->recordKey($id));
192            if ($value === false) {
193                // Indexed but gone - the record expired or was deleted out from
194                // under the set. Drop the stale member rather than carrying it
195                // forever; nothing else prunes it.
196                $this->redis->sRem($this->workerSetKey(), $id);
197                continue;
198            }
199            $record = $this->decode($value);
200            if ($record !== null) {
201                $records[$record->getId()] = $record;
202            }
203        }
204
205        return $records;
206    }
207
208    public function delete(string $id): void
209    {
210        $this->redis->del($this->recordKey($id));
211        $this->redis->sRem($this->workerSetKey(), $id);
212    }
213
214    protected function purgeUndecodable(): int
215    {
216        $this->ensureWorkerSet();
217
218        $removed = 0;
219
220        foreach ($this->redis->sMembers($this->workerSetKey()) as $id) {
221            $value = $this->redis->get($this->recordKey($id));
222            if (($value !== false) && ($this->decode($value) === null)) {
223                $this->delete($id);
224                $removed++;
225            }
226        }
227
228        return $removed;
229    }
230
231}