JobQueue\Infrastructure\RedisQueue

Complexity

Total Complexity

38

Size/Duplication

Total Lines	250
Duplicated Lines	0 %

Test Coverage

Coverage

92.39%

Importance

Changes

0

12 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	3	1
A	add()	0	5	2
A	getList()	0	3	3
A	flush()	0	9	2
C	search()	0	51	12
A	delete()	0	7	1
A	find()	0	7	2
A	restore()	0	13	3
B	updateStatus()	0	31	5
A	getKey()	0	7	3
B	fetch()	0	26	3
A	getTaskList()	0	3	1

<?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
            );

            if (is_null($identifier)) {
                throw new \Exception('Empty list');
            }

            // 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) {
            // Reset non finished tasks to `waiting` status
            if (Status::FINISHED !== (string) $task->getStatus()) {
                $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 ?: '*');
    }
}