Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
91.07%
153 / 168
45.45%
5 / 11
 CRAP
0.00%
0 / 1
Queue
91.07%
153 / 168
45.45%
5 / 11
 66.92
0.00%
0 / 1
 configure
100.00%
57 / 57
100.00%
1 / 1
 17
 writeEnv
94.12%
16 / 17
0.00%
0 / 1
 9.02
 writeConfig
94.44%
17 / 18
0.00%
0 / 1
 6.01
 buildWorker
100.00%
14 / 14
100.00%
1 / 1
 5
 createQueue
85.71%
6 / 7
0.00%
0 / 1
 5.07
 createFileAdapter
75.00%
3 / 4
0.00%
0 / 1
 2.06
 createDatabaseAdapter
84.62%
11 / 13
0.00%
0 / 1
 3.03
 createRedisAdapter
0.00%
0 / 9
0.00%
0 / 1
 6
 clear
100.00%
8 / 8
100.00%
1 / 1
 7
 jobsSummary
100.00%
14 / 14
100.00%
1 / 1
 5
 tasksSummary
100.00%
7 / 7
100.00%
1 / 1
 3
1<?php
2declare(strict_types=1);
3/**
4 * Pop PHP Framework (http://www.popphp.org/)
5 *
6 * @link       https://github.com/popphp/popphp-framework
7 * @author     Nick Sagona, III <dev@noladev.com>
8 * @copyright  Copyright (c) 2009-2027 NOLA Interactive, LLC.
9 * @license    http://www.popphp.org/license     New BSD License
10 */
11
12/**
13 * @namespace
14 */
15namespace Pop\Kettle\Model;
16
17use Pop\Console\Console;
18use Pop\Kettle\Exception;
19use Pop\Utils\AbstractModel;
20
21/**
22 * Queue model class
23 *
24 * @category   Pop\Kettle
25 * @package    Pop\Kettle
26 * @author     Nick Sagona, III <dev@noladev.com>
27 * @copyright  Copyright (c) 2009-2027 NOLA Interactive, LLC.
28 * @license    http://www.popphp.org/license     New BSD License
29 * @version    3.0.0
30 */
31class Queue extends AbstractModel
32{
33
34    /**
35     * Configure queue
36     *
37     * @param  Console $console
38     * @param  string  $location
39     * @param  string  $queue
40     * @return Queue
41     */
42    public function configure(Console $console, string $location, string $queue = 'default'): Queue
43    {
44        $adapterChoices = [];
45        $i              = 1;
46
47        $console->write($i . ': File');
48        $adapterChoices['file'] = $i;
49        $i++;
50
51        $console->write($i . ': Database');
52        $adapterChoices['database'] = $i;
53        $i++;
54
55        if (class_exists('Redis', false)) {
56            $console->write($i . ': Redis');
57            $adapterChoices['redis'] = $i;
58            $i++;
59        }
60
61        $console->write();
62        $selected = $console->prompt('Please select one of the above queue adapters: ', $adapterChoices);
63        $console->write();
64
65        $adapter = array_search($selected, $adapterChoices);
66        $fields  = [];
67
68        if ($adapter == 'file') {
69            $default = 'data/queue/' . $queue;
70            $folder  = $console->prompt('Queue Folder: [' . $default . '] ', null, true);
71            $folder  = ($folder == '') ? $default : $folder;
72
73            if (!file_exists($location . '/' . $folder)) {
74                mkdir($location . '/' . $folder, 0777, true);
75            }
76
77            $fields['folder'] = realpath($location . '/' . $folder);
78        } else if ($adapter == 'database') {
79            $connection = $console->prompt('DB Connection: [default] ', null, true);
80            $connection = ($connection == '') ? 'default' : $connection;
81
82            $defaultTable = ($queue == 'default') ? 'pop_queue' : 'pop_queue_' . $queue;
83            $table        = $console->prompt('Queue Table: [' . $defaultTable . '] ', null, true);
84            $table        = ($table == '') ? $defaultTable : $table;
85
86            $fields['connection'] = $connection;
87            $fields['table']      = $table;
88        } else if ($adapter == 'redis') {
89            $host = $console->prompt('Redis Host: [localhost] ', null, true);
90            $host = ($host == '') ? 'localhost' : $host;
91
92            $port = $console->prompt('Redis Port: [6379] ', null, true);
93            $port = ($port == '') ? '6379' : $port;
94
95            $defaultPrefix = ($queue == 'default') ? 'pop-queue' : 'pop-queue-' . $queue;
96            $prefix        = $console->prompt('Redis Prefix: [' . $defaultPrefix . '] ', null, true);
97            $prefix        = ($prefix == '') ? $defaultPrefix : $prefix;
98
99            $password = $console->prompt('Redis Password: [none] ', null, true);
100
101            $fields['host']     = $host;
102            $fields['port']     = $port;
103            $fields['prefix']   = $prefix;
104            $fields['password'] = $password;
105        }
106
107        $priority = $console->prompt('Queue Priority (FIFO/FILO): [FIFO] ', null, true);
108        $priority = ($priority == '') ? 'FIFO' : strtoupper($priority);
109
110        $lease = $console->prompt('Lease Seconds: [60] ', null, true);
111        $lease = ($lease == '') ? '60' : $lease;
112
113        $weight = $console->prompt('Queue Weight: [0] ', null, true);
114        $weight = ($weight == '') ? '0' : $weight;
115
116        $fields['priority'] = $priority;
117        $fields['lease']    = $lease;
118        $fields['adapter']  = $adapter;
119
120        $this->writeEnv($location, $queue, $fields);
121        $this->writeConfig($location, $queue, $fields, (int)$weight);
122
123        return $this;
124    }
125
126    /**
127     * Write queue env vars, upserting into .env - unprefixed QUEUE_* for the
128     * 'default' queue, QUEUE_<NAME>_* for any other queue name
129     *
130     * @param  string $location
131     * @param  string $queue
132     * @param  array  $fields
133     * @return void
134     */
135    protected function writeEnv(string $location, string $queue, array $fields): void
136    {
137        $prefix  = ($queue == 'default') ? 'QUEUE_' : 'QUEUE_' . strtoupper($queue) . '_';
138        $envFile = $location . '/.env';
139
140        if (!file_exists($envFile)) {
141            copy(__DIR__ . '/../../config/templates/orig.env', $envFile);
142        }
143
144        $env = file_get_contents($envFile);
145
146        foreach ($fields as $key => $value) {
147            $envKey = $prefix . strtoupper($key);
148            $value  = (string)$value;
149
150            // Any value containing a space has to be quoted, or the resulting .env is unparseable
151            if (str_contains($value, ' ') && !str_starts_with($value, '"') && !str_ends_with($value, '"')) {
152                $value = '"' . $value . '"';
153            }
154
155            $pattern = '/^' . preg_quote($envKey, '/') . '=.*$/m';
156            $line    = $envKey . '=' . $value;
157
158            if (($queue == 'default') && preg_match($pattern, $env)) {
159                // Callback form so that any $1/\1 sequences in the value are written verbatim
160                $env = preg_replace_callback($pattern, fn() => $line, $env);
161            } else {
162                $env .= PHP_EOL . $line;
163            }
164        }
165
166        file_put_contents($envFile, $env);
167
168        (\Dotenv\Dotenv::createMutable($location))->safeLoad();
169    }
170
171    /**
172     * Append this queue's config block to app/config/queue.php, creating the
173     * file first if it doesn't exist yet
174     *
175     * @param  string $location
176     * @param  string $queue
177     * @param  array  $fields
178     * @param  int    $weight
179     * @return void
180     */
181    protected function writeConfig(string $location, string $queue, array $fields, int $weight): void
182    {
183        $configFile = $location . '/app/config/queue.php';
184
185        if (!file_exists($configFile)) {
186            if (!file_exists($location . '/app/config')) {
187                mkdir($location . '/app/config', 0777, true);
188            }
189            file_put_contents($configFile, '<?php' . PHP_EOL . PHP_EOL . 'return [' . PHP_EOL . '];' . PHP_EOL);
190        }
191
192        $prefix = ($queue == 'default') ? 'QUEUE_' : 'QUEUE_' . strtoupper($queue) . '_';
193
194        $block  = '    \'' . $queue . '\' => [' . PHP_EOL;
195        $block .= '        \'adapter\'  => $_ENV[\'' . $prefix . 'ADAPTER\'],' . PHP_EOL;
196
197        foreach ($fields as $key => $value) {
198            if ($key == 'adapter') {
199                continue;
200            }
201            $envKey = $prefix . strtoupper($key);
202            $block .= '        \'' . $key . '\' => $_ENV[\'' . $envKey . '\'],' . PHP_EOL;
203        }
204
205        $block .= '        \'weight\'   => ' . $weight . ',' . PHP_EOL;
206        $block .= '    ],' . PHP_EOL;
207
208        $contents = file_get_contents($configFile);
209        $contents = str_replace('];', $block . '];', $contents);
210
211        file_put_contents($configFile, $contents);
212    }
213
214    /**
215     * Build a worker for the given queue (or every configured queue when $queue == 'all')
216     *
217     * @param  string           $location
218     * @param  \Pop\Application $app
219     * @param  string           $queue
220     * @throws Exception
221     * @return \Pop\Queue\Worker
222     */
223    public function buildWorker(string $location, \Pop\Application $app, string $queue = 'default'): \Pop\Queue\Worker
224    {
225        $configFile = $location . '/app/config/queue.php';
226
227        if (!file_exists($configFile)) {
228            throw new Exception('Error: The queue configuration was not found.');
229        }
230
231        $queueConfig = include $configFile;
232
233        if ($queue == 'all') {
234            $names = array_keys($queueConfig);
235        } else {
236            if (!isset($queueConfig[$queue])) {
237                throw new Exception("Error: The queue configuration was not found for '" . $queue . "'.");
238            }
239            $names = [$queue];
240        }
241
242        $worker = \Pop\Queue\Worker::create(null, $app);
243
244        foreach ($names as $name) {
245            $config = $queueConfig[$name];
246            $worker->addQueue($this->createQueue($location, $name, $config), (int)($config['weight'] ?? 0));
247        }
248
249        return $worker;
250    }
251
252    /**
253     * Build a single Queue object from its stored config
254     *
255     * @param  string $location
256     * @param  string $name
257     * @param  array  $config
258     * @throws Exception
259     * @return \Pop\Queue\Queue
260     */
261    protected function createQueue(string $location, string $name, array $config): \Pop\Queue\Queue
262    {
263        $adapter = match ($config['adapter'] ?? null) {
264            'file'     => $this->createFileAdapter($location, $name, $config),
265            'database' => $this->createDatabaseAdapter($location, $config),
266            'redis'    => $this->createRedisAdapter($config),
267            default    => throw new Exception("Error: Unknown queue adapter '" . ($config['adapter'] ?? '') . "'."),
268        };
269
270        return \Pop\Queue\Queue::create($name, $adapter, $config['priority'] ?? null);
271    }
272
273    /**
274     * @param  string $location
275     * @param  string $name
276     * @param  array  $config
277     * @return \Pop\Queue\Adapter\File
278     */
279    protected function createFileAdapter(string $location, string $name, array $config): \Pop\Queue\Adapter\File
280    {
281        $folder = $config['folder'] ?? ($location . '/data/queue/' . $name);
282        if (!file_exists($folder)) {
283            mkdir($folder, 0777, true);
284        }
285
286        return new \Pop\Queue\Adapter\File($folder, $config['priority'] ?? null, (int)($config['lease'] ?? 60));
287    }
288
289    /**
290     * @param  string $location
291     * @param  array  $config
292     * @throws Exception
293     * @return \Pop\Queue\Adapter\Database
294     */
295    protected function createDatabaseAdapter(string $location, array $config): \Pop\Queue\Adapter\Database
296    {
297        $connection   = $config['connection'] ?? 'default';
298        $dbConfigFile = $location . '/app/config/database.php';
299
300        if (!file_exists($dbConfigFile)) {
301            throw new Exception('Error: The database configuration was not found.');
302        }
303
304        $dbConfig = include $dbConfigFile;
305
306        if (!isset($dbConfig[$connection])) {
307            throw new Exception("Error: The database configuration was not found for '" . $connection . "'.");
308        }
309
310        $db = \Pop\Db\Db::connect(
311            $dbConfig[$connection]['adapter'], array_diff_key($dbConfig[$connection], array_flip(['adapter']))
312        );
313
314        return new \Pop\Queue\Adapter\Database(
315            $db, $config['table'] ?? 'pop_queue', $config['priority'] ?? null, (int)($config['lease'] ?? 60)
316        );
317    }
318
319    /**
320     * @param  array $config
321     * @return \Pop\Queue\Adapter\Redis
322     */
323    protected function createRedisAdapter(array $config): \Pop\Queue\Adapter\Redis
324    {
325        $password = !empty($config['password']) ? $config['password'] : null;
326
327        return new \Pop\Queue\Adapter\Redis(
328            $config['host'] ?? 'localhost',
329            $config['port'] ?? 6379,
330            $config['prefix'] ?? 'pop-queue',
331            $config['priority'] ?? null,
332            (int)($config['lease'] ?? 60),
333            $password
334        );
335    }
336
337    /**
338     * Clear jobs/failed jobs/tasks from the given queue (or every configured queue when $queue == 'all')
339     *
340     * @param  \Pop\Queue\Worker $worker
341     * @param  string            $queue
342     * @param  bool              $failed
343     * @param  bool              $tasks
344     * @return void
345     */
346    public function clear(\Pop\Queue\Worker $worker, string $queue, bool $failed = false, bool $tasks = false): void
347    {
348        $names = ($queue == 'all') ? array_keys($worker->getQueues()) : [$queue];
349
350        foreach ($names as $name) {
351            if (!$failed && !$tasks) {
352                $worker->clear($name);
353            }
354            if ($failed) {
355                $worker->clearFailed($name);
356            }
357            if ($tasks) {
358                $worker->clearTasks($name);
359            }
360        }
361    }
362
363    /**
364     * Summarize pending/dead-letter jobs for a queue
365     *
366     * @param  \Pop\Queue\Queue $queue
367     * @return array
368     */
369    public function jobsSummary(\Pop\Queue\Queue $queue): array
370    {
371        $adapter = $queue->getAdapter();
372
373        $summary = [
374            'pending'  => $adapter->count(),
375            'dead'     => $adapter->countDead(),
376            'deadJobs' => [],
377        ];
378
379        if ($summary['dead'] > 0) {
380            foreach ($adapter->getDeadJobs() as $jobId => $job) {
381                $reason = null;
382                if (($job instanceof \Pop\Queue\Process\AbstractJob) && $job->hasFailedMessages()) {
383                    $messages = $job->getFailedMessages();
384                    $reason   = end($messages);
385                }
386                $summary['deadJobs'][$jobId] = $reason;
387            }
388        }
389
390        return $summary;
391    }
392
393    /**
394     * Summarize scheduled tasks for a queue
395     *
396     * @param  \Pop\Queue\Queue $queue
397     * @return array
398     */
399    public function tasksSummary(\Pop\Queue\Queue $queue): array
400    {
401        $summary = [];
402
403        foreach ($queue->getScheduledTasks() as $taskId => $task) {
404            $summary[$taskId] = [
405                'schedule'    => $task->cron()?->getSchedule(),
406                'gracePeriod' => $task->hasGracePeriod() ? $task->getGracePeriod() : null,
407            ];
408        }
409
410        return $summary;
411    }
412
413}