Resque\DataStore

Complexity

Total Complexity

15

Size/Duplication

Total Lines	100
Duplicated Lines	0 %

Test Coverage

Coverage

100%

Importance

Changes

0

15 Methods

Rating	Name	Duplication	Size	Complexity
A	registerWorker()	0	4	1
A	redisKeyForWorkerStartTime()	0	3	1
A	redisKeyForQueue()	0	3	1
A	pushToFailedQueue()	0	3	1
A	setWorkerPayload()	0	3	1
A	unregisterWorker()	0	5	1
A	setNamespace()	0	3	1
A	redisKeyForWorker()	0	3	1
A	workerStarted()	0	4	1
A	setDispatcher()	0	3	1
A	reconnect()	0	4	1
A	popFromQueue()	0	6	1
A	workerDoneWorking()	0	3	1
A	pushToQueue()	0	13	1
A	__construct()	0	4	1

<?php

declare(strict_types=1);

namespace Resque;

use Predis\Client;
use Resque\Dispatchers\Noop;
use Resque\Interfaces\Dispatcher;
use Resque\Tasks\BeforeJobPop;
use Resque\Tasks\BeforeJobPush;

class DataStore
{
    public const REDIS_DATE_FORMAT = 'Y-m-d H:i:s O';

    private $redis;
    private $dispatcher;
    private $namespace = 'resque';

    public function __construct(Client $client)
    {
        $this->redis = $client;
        $this->dispatcher = new Noop();
    }

    public function setDispatcher(Dispatcher $dispatcher): void
    {
        $this->dispatcher = $dispatcher;
    }

    public function setNamespace(string $namespace): void
    {
        $this->namespace = $namespace;
    }

    public function pushToQueue(string $queueName, string $json): void
    {
        $queueKey = $this->redisKeyForQueue($queueName);
        $payload = [
            'queue_key' => $queueKey,
            'queue_name' => $queueName,
            'json' => $json,
            'command' => 'rpush',
        ];
        $payload = $this->dispatcher->dispatch(BeforeJobPush::class, $payload);
        $command = $payload['command'];
        $this->redis->sadd("{$this->namespace}:queues", $payload['queue_name']);
        $this->redis->$command($payload['queue_key'], $payload['json']);
    }

    public function popFromQueue(string $queueName): string
    {
        $payload = ['command' => 'lpop', 'queue_name' => $queueName];
        $payload = $this->dispatcher->dispatch(BeforeJobPop::class, $payload);
        $command = $payload['command'];
        return $this->redis->$command($this->redisKeyForQueue($payload['queue_name']));
    }

    public function redisKeyForQueue(string $queueName): string
    {
        return "{$this->namespace}:queue:{$queueName}";
    }

    public function pushToFailedQueue(string $json): void
    {
        $this->pushToQueue('failed', $json);
    }

    public function registerWorker(string $workerId): void
    {
        $this->redis->sadd("{$this->namespace}:workers", $workerId);
        $this->workerStarted($workerId);
    }

    public function unregisterWorker(string $workerId): void
    {
        $this->redis->srem("{$this->namespace}:workers", $workerId);
        $this->redis->del($this->redisKeyForWorker($workerId));
        $this->redis->del($this->redisKeyForWorkerStartTime($workerId));
    }

    public function workerStarted(string $workerId): void
    {
        $startTime = (new \DateTime())->format(self::REDIS_DATE_FORMAT);
        $this->redis->set($this->redisKeyForWorkerStartTime($workerId), $startTime);
    }

    public function redisKeyForWorker(string $workerId): string
    {
        return "{$this->namespace}:worker:{$workerId}";
    }

    public function redisKeyForWorkerStartTime(string $workerId): string
    {
        return "{$this->redisKeyForWorker($workerId)}:started";
    }

    public function setWorkerPayload(string $workerId, string $data): void
    {
        $this->redis->set($this->redisKeyForWorker($workerId), $data);
    }

    public function workerDoneWorking(string $workerId): void
    {
        $this->redis->del($this->redisKeyForWorker($workerId));
    }

    public function reconnect(): void
    {
        $this->redis->disconnect();
        $this->redis->connect();
    }
}