11125400325

Committed 13 Aug 2024 10:37AM UTC coverage: 78.701%. Remained the same

Build # 11125400325

Build Type

push

github

Committed by

daycry

Commit Message

Fix: upgrade

Run Details

7 of 8 new or added lines in 2 files covered. (87.5%)

2 existing lines in 1 file now uncovered.

303 of 385 relevant lines covered (78.7%)

7.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.05

/src/Commands/QueueWorkerCommand.php

<?php

declare(strict_types=1);

/**
 * This file is part of Daycry Queues.
 *
 * (c) Daycry <daycry9@proton.me>
 *
 * For the full copyright and license information, please view
 * the LICENSE file that was distributed with this source code.
 */

namespace Daycry\Queues\Commands;

use CodeIgniter\CLI\BaseCommand;
use CodeIgniter\CLI\CLI;
use CodeIgniter\HTTP\Response;
use Config\Services;
use Daycry\Exceptions\Interfaces\BaseExceptionInterface;
use Daycry\Queues\Exceptions\QueueException;
use Daycry\Queues\Job;
use Daycry\Queues\Libraries\Utils;
use Exception;

class QueueWorkerCommand extends BaseCommand
{
    protected $group       = 'Queues';
    protected $name        = 'queues:worker';
    protected $description = 'Start queue worker.';
    protected $usage       = 'queue:run <queue> [Options]';
    protected $arguments   = ['queue' => 'The queue name.'];
    protected $options     = ['--oneTime' => 'Only executes one time.'];
    protected bool $locked = false;

    protected function earlyChecks(Job $job): void
    {
    }

    protected function lateChecks(Job $job): void
    {
    }

    protected function earlyCallbackChecks(Job $job): void
    {
    }

    protected function lateCallbackChecks(Job $job): void
    {
    }

    public function run(array $params): void
    {
        $queue = $params[0] ?? CLI::getOption('queue');

        // CLI::write('Queue "'. $queue .'" started successfully.', 'green');

        $oneTime = false;
        if (array_key_exists('oneTime', $params) || CLI::getOption('oneTime')) {
            $oneTime = true;
        }

        // @codeCoverageIgnoreStart
        if (empty($queue)) {
            $queue = CLI::prompt(lang('Queue.insertQueue'));
        }
        // @codeCoverageIgnoreEnd

        while (true) {
            $queues = Utils::parseConfigFile(service('settings')->get('Queue.queues'));

            $response = [];

            Services::resetSingle('request');
            Services::resetSingle('response');

            try {
                $workers = service('settings')->get('Queue.workers');
                $worker  = service('settings')->get('Queue.worker');

                if (! array_key_exists($worker, $workers)) {
                    throw QueueException::forInvalidWorker($worker);
                }

                $worker = new $workers[$worker]();

                $job = $worker->watch($queue);

                if (isset($job)) {
                    $this->locked = true;

                    $dataJob = $worker->getDataJob();
                    $j       = new Job($dataJob);

                    $this->earlyChecks($j);

                    $result = $j->run();

                    $response['status'] = true;

                    if (! $result instanceof Response) {
                        $result = (Services::response(null, true))->setStatusCode(200)->setBody($result);
                    }

                    $response['statusCode'] = $result->getStatusCode();
                    $response['data']       = $result->getBody();

                    $this->lateChecks($j);
                }
            } catch (BaseExceptionInterface $e) {
                $response['statusCode'] = $e->getCode();
                $response['error']      = $e->getMessage();
                $response['status']     = false;
                $worker->removeJob($j, true);
                $this->showError($e);
            } catch (Exception $e) {
                $response['statusCode'] = $e->getCode();
                $response['error']      = $e->getMessage();
                $response['status']     = false;
            }

            if ($response && isset($job)) {
                try {
                    if ($response['status'] === true || $j->getAttempt() >= service('settings')->get('Queue.maxAttempts')) {
                        $worker->removeJob($j, false);
                    }

                    // callback
                    if ($cb = $j->getCallback()) {
                        $cb->options->body = $response;
                        $c                 = new Job();
                        $c->url($cb->url, $cb->options);

                        $this->earlyCallbackChecks($c);
                        $r = $c->run();
                        $this->lateCallbackChecks($c);
                    }
                } catch (BaseExceptionInterface $e) {
                    $this->showError($e);
                }
            }

            $this->locked = false;
            $response     = [];
            unset($j, $job);

            sleep(service('settings')->get('Queue.waitingTimeBetweenJobs'));

            if ($oneTime) {
                return;
            }
        }
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	/**
6	* This file is part of Daycry Queues.
7	*
8	* (c) Daycry <daycry9@proton.me>
9	*
10	* For the full copyright and license information, please view
11	* the LICENSE file that was distributed with this source code.
12	*/
13
14	namespace Daycry\Queues\Commands;
15
16	use CodeIgniter\CLI\BaseCommand;
17	use CodeIgniter\CLI\CLI;
18	use CodeIgniter\HTTP\Response;
19	use Config\Services;
20	use Daycry\Exceptions\Interfaces\BaseExceptionInterface;
21	use Daycry\Queues\Exceptions\QueueException;
22	use Daycry\Queues\Job;
23	use Daycry\Queues\Libraries\Utils;
24	use Exception;
25
26	class QueueWorkerCommand extends BaseCommand
27	{
28	protected $group = 'Queues';
29	protected $name = 'queues:worker';
30	protected $description = 'Start queue worker.';
31	protected $usage = 'queue:run <queue> [Options]';
32	protected $arguments = ['queue' => 'The queue name.'];
33	protected $options = ['--oneTime' => 'Only executes one time.'];
34	protected bool $locked = false;
35
36	protected function earlyChecks(Job $job): void
37	{
38	}	15✔
39
40	protected function lateChecks(Job $job): void
41	{
42	}	15✔
43
44	protected function earlyCallbackChecks(Job $job): void
45	{
46	}	3✔
47
48	protected function lateCallbackChecks(Job $job): void
49	{
50	}	3✔
51
52	public function run(array $params): void
53	{
54	$queue = $params[0] ?? CLI::getOption('queue');	15✔
55
56	// CLI::write('Queue "'. $queue .'" started successfully.', 'green');
57
58	$oneTime = false;	15✔
59	if (array_key_exists('oneTime', $params) \|\| CLI::getOption('oneTime')) {	15✔
60	$oneTime = true;	15✔
61	}
62
63	// @codeCoverageIgnoreStart
64	if (empty($queue)) {	15✔
65	$queue = CLI::prompt(lang('Queue.insertQueue'));	×
66	}
67	// @codeCoverageIgnoreEnd
68
69	while (true) {	15✔
70	$queues = Utils::parseConfigFile(service('settings')->get('Queue.queues'));	15✔
71
72	$response = [];	15✔
73
74	Services::resetSingle('request');	15✔
75	Services::resetSingle('response');	15✔
76
77	try {
78	$workers = service('settings')->get('Queue.workers');	15✔
79	$worker = service('settings')->get('Queue.worker');	15✔
80
81	if (! array_key_exists($worker, $workers)) {	15✔
82	throw QueueException::forInvalidWorker($worker);	×
83	}
84
85	$worker = new $workers[$worker]();	15✔
86
87	$job = $worker->watch($queue);	15✔
88
89	if (isset($job)) {	15✔
90	$this->locked = true;	15✔
91
92	$dataJob = $worker->getDataJob();	15✔
93	$j = new Job($dataJob);	15✔
94
95	$this->earlyChecks($j);	15✔
96
97	$result = $j->run();	15✔
98
99	$response['status'] = true;	15✔
100
101	if (! $result instanceof Response) {	15✔
102	$result = (Services::response(null, true))->setStatusCode(200)->setBody($result);	12✔
103	}
104
105	$response['statusCode'] = $result->getStatusCode();	15✔
106	$response['data'] = $result->getBody();	15✔
107
108	$this->lateChecks($j);	15✔
109	}
110	} catch (BaseExceptionInterface $e) {	×
111	$response['statusCode'] = $e->getCode();	×
112	$response['error'] = $e->getMessage();	×
113	$response['status'] = false;	×
114	$worker->removeJob($j, true);	×
115	$this->showError($e);	×
116	} catch (Exception $e) {	×
117	$response['statusCode'] = $e->getCode();	×
118	$response['error'] = $e->getMessage();	×
119	$response['status'] = false;	×
120	}
121
122	if ($response && isset($job)) {	15✔
123	try {
124	if ($response['status'] === true \|\| $j->getAttempt() >= service('settings')->get('Queue.maxAttempts')) {	15✔
125	$worker->removeJob($j, false);	15✔
126	}
127
128	// callback
129	if ($cb = $j->getCallback()) {	15✔
130	$cb->options->body = $response;	3✔
131	$c = new Job();	3✔
132	$c->url($cb->url, $cb->options);	3✔
133
134	$this->earlyCallbackChecks($c);	3✔
135	$r = $c->run();	3✔
136	$this->lateCallbackChecks($c);	15✔
137	}
138	} catch (BaseExceptionInterface $e) {	×
139	$this->showError($e);	×
140	}
141	}
142
143	$this->locked = false;	15✔
144	$response = [];	15✔
145	unset($j, $job);	15✔
146
147	sleep(service('settings')->get('Queue.waitingTimeBetweenJobs'));	15✔
148
149	if ($oneTime) {	15✔
150	return;	15✔
151	}
152	}
153	}
154	}

daycry / queues / 11125400325

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous