PredisAdapter::readCommand() - Code Metrics - Inspection of "Replaced most operations with lua scripts" - mgdigital/BusQue - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 4c16ee...e71fad )

by Mike

created 2016-06-09 10:28 UTC

PredisAdapter::readCommand() A

↳ Parent: PredisAdapter

Complexity

Conditions	2
Paths	2

Size

Total Lines	8
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
dl	0
loc	8
ccs	0
cts	8
cp	0
rs	9.4285
c	0
b	0
f	0
cc	2
eloc	5
nc	2
nop	2
crap	6

<?php

namespace MGDigital\BusQue\Predis;

use MGDigital\BusQue\Exception\CommandNotFoundException;
use MGDigital\BusQue\Exception\TimeoutException;
use MGDigital\BusQue\QueueAdapterInterface;
use MGDigital\BusQue\ReceivedCommand;
use MGDigital\BusQue\ReceivedScheduledCommand;
use MGDigital\BusQue\SchedulerAdapterInterface;
use MGDigital\BusQue\SchedulerWorker;
use Predis\Client;
use Predis\ClientContextInterface;
use Predis\Connection\ConnectionException;

class PredisAdapter implements QueueAdapterInterface, SchedulerAdapterInterface
{

    const LUA_PATH = __DIR__ . '/../../lua';

    private $client;

    public function __construct(Client $client)
    {
        $this->client = $client;
    }

    public function queueCommand(string $queueName, string $id, string $serialized)
    {
        $this->executeLuaScript('queue_message', [$queueName, $id, $serialized]);
    }

    public function awaitCommand(string $queueName, int $timeout = null): ReceivedCommand
    {
        $stopwatchStart = time();
        $this->client->ping();
        try {
            $id = $this->client->brpoplpush(":{$queueName}:queue", ":{$queueName}:consuming", $timeout ?? 0);
        } catch (ConnectionException $e) {
            $id = null;
        }
        if (!$id) {
            if ($timeout !== null) {
                $timeout = time() - $stopwatchStart - $timeout;
                if ($timeout <= 0) {
                    throw new TimeoutException;
                }
            }
            return $this->awaitCommand($queueName, $timeout);
        }
        /* @var $id string */
        $serialized = $this->executeLuaScript('receive_message', [$queueName, $id]);
        return new ReceivedCommand($queueName, $id, $serialized);
    }

    public function getCommandStatus(string $queueName, string $id): string
    {
        return $this->client->hget(":{$queueName}:statuses", $id) ?? self::STATUS_NOT_FOUND;
    }

    public function setCommandCompleted(string $queueName, string $id)
    {
        $this->executeLuaScript('acknowledge_message', [$queueName, $id]);
    }

    public function setCommandFailed(string $queueName, string $id)
    {
        $this->executeLuaScript('reject_message', [$queueName, $id]);
    }

    public function putQueue(string $queueName)
    {
        $this->client->sadd(':queues', [ $queueName ]);
    }

    public function getQueueNames(): array
    {
        return $this->client->smembers(':queues');
    }

    public function getQueuedCount(string $queueName): int
    {
        return $this->client->llen(":{$queueName}:queue");
    }

    public function getQueuedIds(string $queueName, int $offset = 0, int $limit = 10): array
    {
        return $this->client->lrange(":{$queueName}:queue", $offset, $limit);
    }

    public function getConsumingCount(string $queueName): int
    {
        return $this->client->llen(":{$queueName}:consuming");
    }

    public function getConsumingIds(string $queueName, int $offset = 0, int $limit = 10): array
    {
        return $this->client->lrange(":{$queueName}:consuming", $offset, $limit);
    }

    public function readCommand(string $queueName, string $id): string
    {
        $serialized = $this->client->hget(":{$queueName}:messages", $id);
        if ($serialized === null) {
            throw new CommandNotFoundException();
        }
        return $serialized;
    }

    public function clearQueue(string $queueName)
    {
        $this->executeLuaScript('empty_queue', [$queueName]);
    }

    public function deleteQueue(string $queueName)
    {
        $this->clearQueue($queueName);
        $this->clearSchedule([$queueName]);
    }

    public function purgeCommand(string $queueName, string $id)
    {
        $this->executeLuaScript('purge_message', [$queueName, $id]);
    }

    public function scheduleCommand(string $queueName, string $id, string $serialized, \DateTime $dateTime)
    {
        $this->executeLuaScript('schedule_message', [$queueName, $id, $serialized, $dateTime->getTimestamp()]);
    }

    public function cancelScheduledCommand(string $queueName, string $id)
    {
        $this->purgeCommand($queueName, $id);
    }

    public function clearSchedule(array $queueNames = null, \DateTime $start = null, \DateTime $end = null)
    {
        if ($queueNames === null) {
            $queueNames = [ null ];
        }
        foreach ($queueNames as $queueName) {
            $this->executeLuaScript('clear_schedule', [
                $queueName,
                $start ? $start->getTimestamp() : '-inf',
                $end ? $end->getTimestamp() : '+inf'
            ]);
        }
    }

    public function receiveDueCommands(
        \DateTime $now,
        int $limit = SchedulerWorker::DEFAULT_THROTTLE,
        \DateTime $startTime = null
    ): array {
        if ($startTime === null) {
            $start = 0;
        } else {
            $start = $startTime->getTimestamp();
        }
        
        $results = $this->executeLuaScript('receive_due_messages', [
            $start,
            $now->getTimestamp(),
            $limit
        ]);

        $commands = [ ];

        foreach ($results as $result) {
            list($queueName, $id, $message, $score) = $result;
            $commands[ ] = new ReceivedScheduledCommand(
                $queueName,
                $id,
                $message,
                new \DateTime('@'.$score)
            );
        }
        return $commands;
    }

    private function executeLuaScript(string $script, array $args)
    {
        $command = new LuaFileCommand(static::LUA_PATH . '/' . $script . '.lua');
        $command->setArguments($args);
        return $this->client->executeCommand($command);
    }
}


1			<?php
2
3			namespace MGDigital\BusQue\Predis;
4
5			use MGDigital\BusQue\Exception\CommandNotFoundException;
6			use MGDigital\BusQue\Exception\TimeoutException;
7			use MGDigital\BusQue\QueueAdapterInterface;
8			use MGDigital\BusQue\ReceivedCommand;
9			use MGDigital\BusQue\ReceivedScheduledCommand;
10			use MGDigital\BusQue\SchedulerAdapterInterface;
11			use MGDigital\BusQue\SchedulerWorker;
12			use Predis\Client;
13			use Predis\ClientContextInterface;
14			use Predis\Connection\ConnectionException;
15
16			class PredisAdapter implements QueueAdapterInterface, SchedulerAdapterInterface
17			{
18
19			const LUA_PATH = __DIR__ . '/../../lua';
20
21			private $client;
22
23			public function __construct(Client $client)
24			{
25			$this->client = $client;
26			}
27
28			public function queueCommand(string $queueName, string $id, string $serialized)
29			{
30			$this->executeLuaScript('queue_message', [$queueName, $id, $serialized]);
31			}
32
33			public function awaitCommand(string $queueName, int $timeout = null): ReceivedCommand
34			{
35			$stopwatchStart = time();
36			$this->client->ping();
37			try {
38			$id = $this->client->brpoplpush(":{$queueName}:queue", ":{$queueName}:consuming", $timeout ?? 0);
39			} catch (ConnectionException $e) {
40			$id = null;
41			}
42			if (!$id) {
43			if ($timeout !== null) {
44			$timeout = time() - $stopwatchStart - $timeout;
45			if ($timeout <= 0) {
46			throw new TimeoutException;
47			}
48			}
49			return $this->awaitCommand($queueName, $timeout);
50			}
51			/* @var $id string */
52			$serialized = $this->executeLuaScript('receive_message', [$queueName, $id]);
53			return new ReceivedCommand($queueName, $id, $serialized);
54			}
55
56			public function getCommandStatus(string $queueName, string $id): string
57			{
58			return $this->client->hget(":{$queueName}:statuses", $id) ?? self::STATUS_NOT_FOUND;
59			}
60
61			public function setCommandCompleted(string $queueName, string $id)
62			{
63			$this->executeLuaScript('acknowledge_message', [$queueName, $id]);
64			}
65
66			public function setCommandFailed(string $queueName, string $id)
67			{
68			$this->executeLuaScript('reject_message', [$queueName, $id]);
69			}
70
71			public function putQueue(string $queueName)
72			{
73			$this->client->sadd(':queues', [ $queueName ]);
74			}
75
76			public function getQueueNames(): array
77			{
78			return $this->client->smembers(':queues');
79			}
80
81			public function getQueuedCount(string $queueName): int
82			{
83			return $this->client->llen(":{$queueName}:queue");
84			}
85
86			public function getQueuedIds(string $queueName, int $offset = 0, int $limit = 10): array
87			{
88			return $this->client->lrange(":{$queueName}:queue", $offset, $limit);
89			}
90
91			public function getConsumingCount(string $queueName): int
92			{
93			return $this->client->llen(":{$queueName}:consuming");
94			}
95
96			public function getConsumingIds(string $queueName, int $offset = 0, int $limit = 10): array
97			{
98			return $this->client->lrange(":{$queueName}:consuming", $offset, $limit);
99			}
100
101			public function readCommand(string $queueName, string $id): string
102			{
103			$serialized = $this->client->hget(":{$queueName}:messages", $id);
104			if ($serialized === null) {
105			throw new CommandNotFoundException();
106			}
107			return $serialized;
108			}
109
110			public function clearQueue(string $queueName)
111			{
112			$this->executeLuaScript('empty_queue', [$queueName]);
113			}
114
115			public function deleteQueue(string $queueName)
116			{
117			$this->clearQueue($queueName);
118			$this->clearSchedule([$queueName]);
119			}
120
121			public function purgeCommand(string $queueName, string $id)
122			{
123			$this->executeLuaScript('purge_message', [$queueName, $id]);
124			}
125
126			public function scheduleCommand(string $queueName, string $id, string $serialized, \DateTime $dateTime)
127			{
128			$this->executeLuaScript('schedule_message', [$queueName, $id, $serialized, $dateTime->getTimestamp()]);
129			}
130
131			public function cancelScheduledCommand(string $queueName, string $id)
132			{
133			$this->purgeCommand($queueName, $id);
134			}
135
136			public function clearSchedule(array $queueNames = null, \DateTime $start = null, \DateTime $end = null)
137			{
138			if ($queueNames === null) {
139			$queueNames = [ null ];
140			}
141			foreach ($queueNames as $queueName) {
142			$this->executeLuaScript('clear_schedule', [
143			$queueName,
144			$start ? $start->getTimestamp() : '-inf',
145			$end ? $end->getTimestamp() : '+inf'
146			]);
147			}
148			}
149
150			public function receiveDueCommands(
151			\DateTime $now,
152			int $limit = SchedulerWorker::DEFAULT_THROTTLE,
153			\DateTime $startTime = null
154			): array {
155			if ($startTime === null) {
156			$start = 0;
157			} else {
158			$start = $startTime->getTimestamp();
159			}
160
161			$results = $this->executeLuaScript('receive_due_messages', [
162			$start,
163			$now->getTimestamp(),
164			$limit
165			]);
166
167			$commands = [ ];
168
169			foreach ($results as $result) {
170			list($queueName, $id, $message, $score) = $result;
171			$commands[ ] = new ReceivedScheduledCommand(
172			$queueName,
173			$id,
174			$message,
175			new \DateTime('@'.$score)
176			);
177			}
178			return $commands;
179			}
180
181			private function executeLuaScript(string $script, array $args)
182			{
183			$command = new LuaFileCommand(static::LUA_PATH . '/' . $script . '.lua');
184			$command->setArguments($args);
185			return $this->client->executeCommand($command);
186			}
187			}
188

mgdigital / BusQue

Push — master ( 4c16ee...e71fad )

PredisAdapter::readCommand() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like