RedisQueue::getTaskList() - Code Metrics - Inspection of "Rollback on env setting from webservice tests" - bricev/JobQueue - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — v2 ( 9bff46...d4d6f5 )

by Brice

created 2018-03-07 14:07 UTC

RedisQueue::getTaskList() A

↳ Parent: RedisQueue

Complexity

Conditions	1
Paths	1

Size

Total Lines	3
Code Lines	1

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	1
nc	1
nop	1
dl	0
loc	3
ccs	2
cts	2
cp	1
crap	1
rs	10
c	0
b	0
f	0

<?php

namespace JobQueue\Infrastructure;

use JobQueue\Domain\Task\Profile;
use JobQueue\Domain\Task\Queue;
use JobQueue\Domain\Task\Status;
use JobQueue\Domain\Task\Task;
use Predis\Client;
use Predis\Collection\Iterator\Keyspace;

/**
 * Redis implementation of the task queue.
 *
 * Used keys:
 * - jobqueue.tasks.{task_identifier}
 * - jobqueue.{profile}.waiting
 * - jobqueue.{profile}.running
 * - jobqueue.{profile}.failed
 */
final class RedisQueue implements Queue
{
    /**
     *
     * @var Client
     */
    private $predis;

    /**
     *
     * @param Client $predis
     */
    public function __construct(Client $predis)
    {
        $this->predis = $predis;
    }

    /**
     *
     * @param Task[] ...$tasks
     */
    public function add(Task ...$tasks)
    {
        foreach ($tasks as $task) {
            $this->predis->set($this->getKey($task), serialize($task));
            $this->predis->lpush($this->getTaskList($task), $task->getIdentifier());
        }
    }

    /**
     *
     * @param Profile $profile
     * @return Task
     */
    public function fetch(Profile $profile): Task
    {
        try {
            $identifier = $this->predis->brpoplpush(
                $this->getList($profile, new Status(Status::WAITING)),
                $this->getList($profile, new Status(Status::RUNNING)),
                300 // 5 minutes
            );

            // Find and update task
            $task = $this->find($identifier);
            $task->updateStatus(new Status(Status::RUNNING));
            $this->predis->set($this->getKey($task), serialize($task));

            return $task;

        } catch (\Exception $e) {
            // sleep a little to avoid high CPU consuming infinite loops...
            sleep(3);

            // ... and try again
            return $this->fetch($profile);
        }
    }

    /**
     *
     * @param string $identifier
     * @return Task
     */
    public function find(string $identifier): Task
    {
        if (!$serializedTask = $this->predis->get($this->getKey($identifier))) {
            throw new \RuntimeException(sprintf('Task %s does not exists', $identifier));
        }

        return unserialize($serializedTask);
    }

    /**
     *
     * @param Task   $task
     * @param Status $status
     */
    public function updateStatus(Task $task, Status $status)
    {
        if ((string) $status === (string) $task->getStatus()) {
            return;
        }

/**
 * Can be useful to mark tasks as success after a crash
 * Required now as the worker needs to mark tasks as finished
 */
//        if (Status::RUNNING === (string) $task->getStatus()) {
//            throw new \RuntimeException('The status of a running task can\'t be updated');
//        }

        if (Status::RUNNING === (string) $status) {
            throw new \RuntimeException('A task can not be marked as "running"');
        }

        // Remove task from its current list (there is no `finished` list)
        if (Status::FINISHED !== (string) $task->getStatus()) {
            $this->predis->lrem($this->getTaskList($task), 0, (string) $task);
        }

        // Insert the task to its new list (except for `finished` tasks)
        if (Status::FINISHED !== (string) $status) {
            $this->predis->lpush($this->getList($task->getProfile(), $status), (string) $task);
        }

        // Update task
        $task->updateStatus($status);
        $this->predis->set($this->getKey($task), serialize($task));
    }

    /**
     *
     * @param Profile|null $profile
     * @param Status|null  $status
     * @param array        $tags
     * @param string       $orderBy
     * @return Task[]
     * @throws \Exception
     */
    public function search(Profile $profile = null, Status $status = null, array $tags = [], string $orderBy = 'date'): array
    {
        if (!in_array($orderBy, ['date', 'profile', 'status'])) {
            throw new \Exception(sprintf('Impossible to order by "%s"', $orderBy));
        }

        // List all tasks
        $tasks = [];
        foreach (new Keyspace($this->predis, $this->getKey()) as $key) {

            $task = unserialize($this->predis->get($key)); /** @var Task $task */

            if ($profile and (string) $profile !== (string) $task->getProfile()) {
                continue;
            }

            if ($status and (string) $status !== (string) $task->getStatus()) {
                continue;
            }

            if (!empty($tags)) {
                // Check that the task has at least one tag
                if (empty(array_intersect($tags, $task->getTags()))) {
                    continue;
                }
            }

            $tasks[] = [
                'date' => $task->getCreatedAt(),
                'profile' => (string) $task->getProfile(),
                'status' => array_search((string) $task->getStatus(), Status::listStatus()),
                'task' => $task,
            ];
        }

        // Order Tasks
        usort($tasks, function ($a, $b) use ($orderBy) {
            $aValue = $a[$orderBy];
            $bValue = $b[$orderBy];

            if ($aValue === $bValue) return 0;

            return $aValue < $bValue ? -1 : 1;
        });

        // Clean return
        foreach ($tasks as $key => $task) {
            $tasks[$key] = $task['task'];
        }

        return $tasks;
    }

    /**
     *
     * @param string $identifier
     */
    public function delete(string $identifier)
    {
        $task = $this->find($identifier);

        $this->predis->lrem($this->getTaskList($task), 0, (string) $task);

        $this->predis->del($this->getKey($task));
    }

    public function flush()
    {
        $pipe = $this->predis->pipeline();

        foreach ($this->predis->keys('jobqueue.*') as $key) {
            $pipe->del($key);
        }

        $pipe->execute();
    }

    public function restore()
    {
        $tasks = $this->search();

        $this->flush();

        foreach ($tasks as $task) {
            $task->updateStatus(new Status(Status::WAITING));
            $this->add($task);
        }
    }

    /**
     *
     * @param Profile $profile
     * @param Status  $status
     * @return string
     */
    protected function getList(Profile $profile = null, Status $status = null): string
    {
        return sprintf('jobqueue.%s.%s', $profile ?: '*', $status ?: '*');
    }

    /**
     *
     * @param Task $task
     * @return string
     */
    protected function getTaskList(Task $task): string
    {
        return $this->getList($task->getProfile(), $task->getStatus());
    }

    /**
     *
     * @param $identifier
     * @return string
     */
    protected function getKey($identifier = null): string
    {
        if ($identifier instanceof Task) {
            $identifier = (string) $identifier->getIdentifier();
        }

        return sprintf('jobqueue.tasks.%s', $identifier ?: '*');
    }
}

1		<?php
2
3		namespace JobQueue\Infrastructure;
4
5		use JobQueue\Domain\Task\Profile;
6		use JobQueue\Domain\Task\Queue;
7		use JobQueue\Domain\Task\Status;
8		use JobQueue\Domain\Task\Task;
9		use Predis\Client;
10		use Predis\Collection\Iterator\Keyspace;
11
12		/**
13		* Redis implementation of the task queue.
14		*
15		* Used keys:
16		* - jobqueue.tasks.{task_identifier}
17		* - jobqueue.{profile}.waiting
18		* - jobqueue.{profile}.running
19		* - jobqueue.{profile}.failed
20		*/
21		final class RedisQueue implements Queue
22		{
23		/**
24		*
25		* @var Client
26		*/
27		private $predis;
28
29		/**
30		*
31		* @param Client $predis
32		*/
33	2	public function __construct(Client $predis)
34		{
35	2	$this->predis = $predis;
36	2	}
37
38		/**
39		*
40		* @param Task[] ...$tasks
41		*/
42	5	public function add(Task ...$tasks)
43		{
44	5	foreach ($tasks as $task) {
45	5	$this->predis->set($this->getKey($task), serialize($task));
46	5	$this->predis->lpush($this->getTaskList($task), $task->getIdentifier());
47		}
48	5	}
49
50		/**
51		*
52		* @param Profile $profile
53		* @return Task
54		*/
55	6	public function fetch(Profile $profile): Task
56		{
57		try {
58	6	$identifier = $this->predis->brpoplpush(
59	6	$this->getList($profile, new Status(Status::WAITING)),
60	6	$this->getList($profile, new Status(Status::RUNNING)),
61	6	300 // 5 minutes
62		);
63
64		// Find and update task
65	6	$task = $this->find($identifier);
66	6	$task->updateStatus(new Status(Status::RUNNING));
67	6	$this->predis->set($this->getKey($task), serialize($task));
68
69	6	return $task;
70
71		} catch (\Exception $e) {
72		// sleep a little to avoid high CPU consuming infinite loops...
73		sleep(3);
74
75		// ... and try again
76		return $this->fetch($profile);
77		}
78		}
79
80		/**
81		*
82		* @param string $identifier
83		* @return Task
84		*/
85	10	public function find(string $identifier): Task
86		{
87	10	if (!$serializedTask = $this->predis->get($this->getKey($identifier))) {
88	1	throw new \RuntimeException(sprintf('Task %s does not exists', $identifier));
89		}
90
91	10	return unserialize($serializedTask);
92		}
93
94		/**
95		*
96		* @param Task $task
97		* @param Status $status
98		*/
99	6	public function updateStatus(Task $task, Status $status)
100		{
101	6	if ((string) $status === (string) $task->getStatus()) {
102		return;
103		}
104
105		/**
106		* Can be useful to mark tasks as success after a crash
107		* Required now as the worker needs to mark tasks as finished
108		*/
109		// if (Status::RUNNING === (string) $task->getStatus()) {
110		// throw new \RuntimeException('The status of a running task can\'t be updated');
111		// }
112
113	6	if (Status::RUNNING === (string) $status) {
114		throw new \RuntimeException('A task can not be marked as "running"');
115		}
116
117		// Remove task from its current list (there is no `finished` list)
118	6	if (Status::FINISHED !== (string) $task->getStatus()) {
119	6	$this->predis->lrem($this->getTaskList($task), 0, (string) $task);
120		}
121
122		// Insert the task to its new list (except for `finished` tasks)
123	6	if (Status::FINISHED !== (string) $status) {
124	2	$this->predis->lpush($this->getList($task->getProfile(), $status), (string) $task);
125		}
126
127		// Update task
128	6	$task->updateStatus($status);
129	6	$this->predis->set($this->getKey($task), serialize($task));
130	6	}
131
132		/**
133		*
134		* @param Profile\|null $profile
135		* @param Status\|null $status
136		* @param array $tags
137		* @param string $orderBy
138		* @return Task[]
139		* @throws \Exception
140		*/
141	9	public function search(Profile $profile = null, Status $status = null, array $tags = [], string $orderBy = 'date'): array
142		{
143	9	if (!in_array($orderBy, ['date', 'profile', 'status'])) {
144		throw new \Exception(sprintf('Impossible to order by "%s"', $orderBy));
145		}
146
147		// List all tasks
148	9	$tasks = [];
149	9	foreach (new Keyspace($this->predis, $this->getKey()) as $key) {
150
151	7	$task = unserialize($this->predis->get($key)); /** @var Task $task */
152
153	7	if ($profile and (string) $profile !== (string) $task->getProfile()) {
154	2	continue;
155		}
156
157	7	if ($status and (string) $status !== (string) $task->getStatus()) {
158	1	continue;
159		}
160
161	7	if (!empty($tags)) {
162		// Check that the task has at least one tag
163	2	if (empty(array_intersect($tags, $task->getTags()))) {
164	2	continue;
165		}
166		}
167
168	7	$tasks[] = [
169	7	'date' => $task->getCreatedAt(),
170	7	'profile' => (string) $task->getProfile(),
171	7	'status' => array_search((string) $task->getStatus(), Status::listStatus()),
172	7	'task' => $task,
173		];
174		}
175
176		// Order Tasks
177	9	usort($tasks, function ($a, $b) use ($orderBy) {
178	6	$aValue = $a[$orderBy];
179	6	$bValue = $b[$orderBy];
180
181	6	if ($aValue === $bValue) return 0;
182
183		return $aValue < $bValue ? -1 : 1;
184	9	});
185
186		// Clean return
187	9	foreach ($tasks as $key => $task) {
188	7	$tasks[$key] = $task['task'];
189		}
190
191	9	return $tasks;
192		}
193
194		/**
195		*
196		* @param string $identifier
197		*/
198	1	public function delete(string $identifier)
199		{
200	1	$task = $this->find($identifier);
201
202	1	$this->predis->lrem($this->getTaskList($task), 0, (string) $task);
203
204	1	$this->predis->del($this->getKey($task));
205	1	}
206
207	4	public function flush()
208		{
209	4	$pipe = $this->predis->pipeline();
210
211	4	foreach ($this->predis->keys('jobqueue.*') as $key) {
212	4	$pipe->del($key);
213		}
214
215	4	$pipe->execute();
216	4	}
217
218	2	public function restore()
219		{
220	2	$tasks = $this->search();
221
222	2	$this->flush();
223
224	2	foreach ($tasks as $task) {
225	2	$task->updateStatus(new Status(Status::WAITING));
226	2	$this->add($task);
227		}
228	2	}
229
230		/**
231		*
232		* @param Profile $profile
233		* @param Status $status
234		* @return string
235		*/
236	13	protected function getList(Profile $profile = null, Status $status = null): string
237		{
238	13	return sprintf('jobqueue.%s.%s', $profile ?: '', $status ?: '');
239		}
240
241		/**
242		*
243		* @param Task $task
244		* @return string
245		*/
246	12	protected function getTaskList(Task $task): string
247		{
248	12	return $this->getList($task->getProfile(), $task->getStatus());
249		}
250
251		/**
252		*
253		* @param $identifier
254		* @return string
255		*/
256	20	protected function getKey($identifier = null): string
257		{
258	20	if ($identifier instanceof Task) {
259	13	$identifier = (string) $identifier->getIdentifier();
260		}
261
262	20	return sprintf('jobqueue.tasks.%s', $identifier ?: '*');
263		}
264		}

bricev / JobQueue

Push — v2 ( 9bff46...d4d6f5 )

RedisQueue::getTaskList() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like