JobManager::prioritySave() - Code Metrics - Inspection of "Fix a bug with the loop code" - mmucklo/DtcQueueBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f30a66...783f1c )

by Matthew

created 2017-11-11 22:23 UTC

JobManager::prioritySave() A

↳ Parent: JobManager

Complexity

Conditions	2
Paths	2

Size

Total Lines	18
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	2

Importance

Changes

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

<?php

namespace Dtc\QueueBundle\RabbitMQ;

use Dtc\QueueBundle\Model\PriorityJobManager;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Message\AMQPMessage;

class JobManager extends PriorityJobManager
{
    /** @var AMQPChannel */
    protected $channel;

    /** @var AbstractConnection */
    protected $connection;
    protected $queueArgs;
    protected $exchangeArgs;

    protected $channelSetup = false;

    protected $hostname;
    protected $pid;

    public function __construct()
    {
        $this->hostname = gethostname() ?: '';
        $this->pid = getmypid();
    }

    /**
     * @param string $exchange
     * @param string $type
     * @param bool   $passive
     * @param bool   $durable
     * @param bool   $autoDelete
     */
    public function setExchangeArgs($exchange, $type, $passive, $durable, $autoDelete)
    {
        $this->exchangeArgs = [$exchange, $type, $passive, $durable, $autoDelete];
    }

    /**
     * @param string $queue
     * @param bool   $passive
     * @param bool   $durable
     * @param bool   $exclusive
     * @param bool   $autoDelete
     */
    public function setQueueArgs($queue, $passive, $durable, $exclusive, $autoDelete)
    {
        $arguments = [$queue, $passive, $durable, $exclusive, $autoDelete];

        $this->queueArgs = $arguments;
        if (!ctype_digit(strval($this->maxPriority))) {
            throw new \Exception('Max Priority ('.$this->maxPriority.') needs to be a non-negative integer');
        }
        if (strval(intval($this->maxPriority)) !== strval($this->maxPriority)) {
            throw new \Exception('Priority is higher than '.PHP_INT_MAX);
        }
    }

    public function setAMQPConnection(AbstractConnection $connection)
    {
        $this->connection = $connection;
        $this->channel = $connection->channel();
    }

    /**
     * @return AMQPChannel
     */
    public function getChannel()
    {
        return $this->channel;
    }

    protected function checkChannelArgs()
    {
        if (empty($this->queueArgs)) {
            throw new \Exception(__METHOD__.': queue args need to be set via setQueueArgs(...)');
        }
        if (empty($this->exchangeArgs)) {
            throw new \Exception(__METHOD__.': exchange args need to be set via setExchangeArgs(...)');
        }
    }

    protected function performChannelSetup()
    {
        call_user_func_array([$this->channel, 'exchange_declare'], $this->exchangeArgs);
        if ($this->maxPriority) {
            array_push($this->queueArgs, false);
            array_push($this->queueArgs, ['x-max-priority' => ['I', intval($this->maxPriority)]]);
        }
        call_user_func_array([$this->channel, 'queue_declare'], $this->queueArgs);
        $this->channel->queue_bind($this->queueArgs[0], $this->exchangeArgs[0]);
    }

    public function setupChannel()
    {
        $this->checkChannelArgs();

        if (!$this->channelSetup) {
            $this->performChannelSetup();
            $this->channelSetup = true;
        }
    }

    /**
     * @param \Dtc\QueueBundle\Model\Job $job
     *
     * @return \Dtc\QueueBundle\Model\Job
     *
     * @throws \Exception
     */
    public function prioritySave(\Dtc\QueueBundle\Model\Job $job)
    {
        if (!$job instanceof Job) {
            throw new \Exception('Must be derived from '.Job::class);
        }

        $this->setupChannel();

        $this->validateSaveable($job);
        $this->setJobId($job);

        $msg = new AMQPMessage($job->toMessage());
        $this->setMsgPriority($msg, $job);

        $this->channel->basic_publish($msg, $this->exchangeArgs[0]);

        return $job;
    }

    /**
     * Attach a unique id to a job since RabbitMQ will not.
     *
     * @param \Dtc\QueueBundle\Model\Job $job
     */
    protected function setJobId(\Dtc\QueueBundle\Model\Job $job)
    {
        if (!$job->getId()) {
            $job->setId(uniqid($this->hostname.'-'.$this->pid, true));
        }
    }

    /**
     * Sets the priority of the AMQPMessage.
     *
     * @param AMQPMessage                $msg
     * @param \Dtc\QueueBundle\Model\Job $job
     */
    protected function setMsgPriority(AMQPMessage $msg, \Dtc\QueueBundle\Model\Job $job)
    {
        if (null !== $this->maxPriority) {
            $priority = $job->getPriority();
            $msg->set('priority', $priority);
        }
    }

    protected function calculatePriority($priority)
    {
        $priority = parent::calculatePriority($priority);
        if (null === $priority) {
            return 0;
        }

        return $priority;
    }

    /**
     * @param \Dtc\QueueBundle\Model\Job $job
     *
     * @throws \Exception
     */
    protected function validateSaveable(\Dtc\QueueBundle\Model\Job $job)
    {
        if (null !== $job->getPriority() && null === $this->maxPriority) {
            throw new \Exception('This queue does not support priorities');
        }

        if (!$job instanceof Job) {
            throw new \Exception('Job needs to be instance of '.Job::class);
        }
    }

    protected function verifyGetJobArgs($workerName = null, $methodName = null, $prioritize = true)
    {
        if (null !== $workerName || null !== $methodName || true !== $prioritize) {
            throw new \Exception('Unsupported');
        }
    }

    /**
     * @param string $workerName
     */
    public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
    {
        $this->verifyGetJobArgs($workerName, $methodName, $prioritize);
        $this->setupChannel();

        do {
            $expiredJob = false;
            $job = $this->findJob($expiredJob, $runId);
        } while ($expiredJob);

        return $job;
    }

    /**
     * @param bool $expiredJob
     * @param $runId
     *
     * @return Job|null
     */
    protected function findJob(&$expiredJob, $runId)
    {
        $message = $this->channel->basic_get($this->queueArgs[0]);
        if ($message) {
            $job = new Job();
            $job->fromMessage($message->body);
            $job->setRunId($runId);

            if (($expiresAt = $job->getExpiresAt()) && $expiresAt->getTimestamp() < time()) {
                $expiredJob = true;
                $this->channel->basic_nack($message->delivery_info['delivery_tag']);

                return null;
            }
            $job->setDeliveryTag($message->delivery_info['delivery_tag']);

            return $job;
        }

        return null;
    }

    // Save History get called upon completion of the job
    public function saveHistory(\Dtc\QueueBundle\Model\Job $job)
    {
        if (!$job instanceof Job) {
            throw new \Exception("Expected \Dtc\QueueBundle\RabbitMQ\Job, got ".get_class($job));
        }
        $deliveryTag = $job->getDeliveryTag();
        $this->channel->basic_ack($deliveryTag);

        return;
    }

    public function __destruct()
    {
        if (null !== $this->channel) {
            $this->channel->close();
        }
    }
}


1		<?php
2
3		namespace Dtc\QueueBundle\RabbitMQ;
4
5		use Dtc\QueueBundle\Model\PriorityJobManager;
6		use PhpAmqpLib\Channel\AMQPChannel;
7		use PhpAmqpLib\Connection\AbstractConnection;
8		use PhpAmqpLib\Message\AMQPMessage;
9
10		class JobManager extends PriorityJobManager
11		{
12		/** @var AMQPChannel */
13		protected $channel;
14
15		/** @var AbstractConnection */
16		protected $connection;
17		protected $queueArgs;
18		protected $exchangeArgs;
19
20		protected $channelSetup = false;
21
22		protected $hostname;
23		protected $pid;
24
25	2	public function __construct()
26		{
27	2	$this->hostname = gethostname() ?: '';
28	2	$this->pid = getmypid();
29	2	}
30
31		/**
32		* @param string $exchange
33		* @param string $type
34		* @param bool $passive
35		* @param bool $durable
36		* @param bool $autoDelete
37		*/
38	1	public function setExchangeArgs($exchange, $type, $passive, $durable, $autoDelete)
39		{
40	1	$this->exchangeArgs = [$exchange, $type, $passive, $durable, $autoDelete];
41	1	}
42
43		/**
44		* @param string $queue
45		* @param bool $passive
46		* @param bool $durable
47		* @param bool $exclusive
48		* @param bool $autoDelete
49		*/
50	1	public function setQueueArgs($queue, $passive, $durable, $exclusive, $autoDelete)
51		{
52	1	$arguments = [$queue, $passive, $durable, $exclusive, $autoDelete];
53
54	1	$this->queueArgs = $arguments;
55	1	if (!ctype_digit(strval($this->maxPriority))) {
56	1	throw new \Exception('Max Priority ('.$this->maxPriority.') needs to be a non-negative integer');
57		}
58	1	if (strval(intval($this->maxPriority)) !== strval($this->maxPriority)) {
59		throw new \Exception('Priority is higher than '.PHP_INT_MAX);
60		}
61	1	}
62
63	1	public function setAMQPConnection(AbstractConnection $connection)
64		{
65	1	$this->connection = $connection;
66	1	$this->channel = $connection->channel();
67	1	}
68
69		/**
70		* @return AMQPChannel
71		*/
72		public function getChannel()
73		{
74		return $this->channel;
75		}
76
77	7	protected function checkChannelArgs()
78		{
79	7	if (empty($this->queueArgs)) {
80	1	throw new \Exception(__METHOD__.': queue args need to be set via setQueueArgs(...)');
81		}
82	7	if (empty($this->exchangeArgs)) {
83	1	throw new \Exception(__METHOD__.': exchange args need to be set via setExchangeArgs(...)');
84		}
85	7	}
86
87	1	protected function performChannelSetup()
88		{
89	1	call_user_func_array([$this->channel, 'exchange_declare'], $this->exchangeArgs);
90	1	if ($this->maxPriority) {
91	1	array_push($this->queueArgs, false);
92	1	array_push($this->queueArgs, ['x-max-priority' => ['I', intval($this->maxPriority)]]);
93		}
94	1	call_user_func_array([$this->channel, 'queue_declare'], $this->queueArgs);
95	1	$this->channel->queue_bind($this->queueArgs[0], $this->exchangeArgs[0]);
96	1	}
97
98	7	public function setupChannel()
99		{
100	7	$this->checkChannelArgs();
101
102	7	if (!$this->channelSetup) {
103	1	$this->performChannelSetup();
104	1	$this->channelSetup = true;
105		}
106	7	}
107
108		/**
109		* @param \Dtc\QueueBundle\Model\Job $job
110		*
111		* @return \Dtc\QueueBundle\Model\Job
112		*
113		* @throws \Exception
114		*/
115	6	public function prioritySave(\Dtc\QueueBundle\Model\Job $job)
116		{
117	6	if (!$job instanceof Job) {
118	1	throw new \Exception('Must be derived from '.Job::class);
119		}
120
121	6	$this->setupChannel();
122
123	6	$this->validateSaveable($job);
124	6	$this->setJobId($job);
125
126	6	$msg = new AMQPMessage($job->toMessage());
127	6	$this->setMsgPriority($msg, $job);
128
129	6	$this->channel->basic_publish($msg, $this->exchangeArgs[0]);
130
131	6	return $job;
132		}
133
134		/**
135		* Attach a unique id to a job since RabbitMQ will not.
136		*
137		* @param \Dtc\QueueBundle\Model\Job $job
138		*/
139	6	protected function setJobId(\Dtc\QueueBundle\Model\Job $job)
140		{
141	6	if (!$job->getId()) {
142	6	$job->setId(uniqid($this->hostname.'-'.$this->pid, true));
143		}
144	6	}
145
146		/**
147		* Sets the priority of the AMQPMessage.
148		*
149		* @param AMQPMessage $msg
150		* @param \Dtc\QueueBundle\Model\Job $job
151		*/
152	6	protected function setMsgPriority(AMQPMessage $msg, \Dtc\QueueBundle\Model\Job $job)
153		{
154	6	if (null !== $this->maxPriority) {
155	6	$priority = $job->getPriority();
156	6	$msg->set('priority', $priority);
157		}
158	6	}
159
160	6	protected function calculatePriority($priority)
161		{
162	6	$priority = parent::calculatePriority($priority);
163	6	if (null === $priority) {
164	5	return 0;
165		}
166
167	3	return $priority;
168		}
169
170		/**
171		* @param \Dtc\QueueBundle\Model\Job $job
172		*
173		* @throws \Exception
174		*/
175	6	protected function validateSaveable(\Dtc\QueueBundle\Model\Job $job)
176		{
177	6	if (null !== $job->getPriority() && null === $this->maxPriority) {
178		throw new \Exception('This queue does not support priorities');
179		}
180
181	6	if (!$job instanceof Job) {
182		throw new \Exception('Job needs to be instance of '.Job::class);
183		}
184	6	}
185
186	7	protected function verifyGetJobArgs($workerName = null, $methodName = null, $prioritize = true)
187		{
188	7	if (null !== $workerName \|\| null !== $methodName \|\| true !== $prioritize) {
189	2	throw new \Exception('Unsupported');
190		}
191	6	}
192
193		/**
194		* @param string $workerName
195		*/
196	7	public function getJob($workerName = null, $methodName = null, $prioritize = true, $runId = null)
197		{
198	7	$this->verifyGetJobArgs($workerName, $methodName, $prioritize);
199	6	$this->setupChannel();
200
201		do {
202	6	$expiredJob = false;
203	6	$job = $this->findJob($expiredJob, $runId);
204	6	} while ($expiredJob);
205
206	6	return $job;
207		}
208
209		/**
210		* @param bool $expiredJob
211		* @param $runId
212		*
213		* @return Job\|null
214		*/
215	6	protected function findJob(&$expiredJob, $runId)
216		{
217	6	$message = $this->channel->basic_get($this->queueArgs[0]);
218	6	if ($message) {
219	6	$job = new Job();
220	6	$job->fromMessage($message->body);
221	6	$job->setRunId($runId);
222
223	6	if (($expiresAt = $job->getExpiresAt()) && $expiresAt->getTimestamp() < time()) {
224	1	$expiredJob = true;
225	1	$this->channel->basic_nack($message->delivery_info['delivery_tag']);
226
227	1	return null;
228		}
229	6	$job->setDeliveryTag($message->delivery_info['delivery_tag']);
230
231	6	return $job;
232		}
233
234	3	return null;
235		}
236
237		// Save History get called upon completion of the job
238	1	public function saveHistory(\Dtc\QueueBundle\Model\Job $job)
239		{
240	1	if (!$job instanceof Job) {
241	1	throw new \Exception("Expected \Dtc\QueueBundle\RabbitMQ\Job, got ".get_class($job));
242		}
243	1	$deliveryTag = $job->getDeliveryTag();
244	1	$this->channel->basic_ack($deliveryTag);
245
246	1	return;
247		}
248
249	3	public function __destruct()
250		{
251	3	if (null !== $this->channel) {
252	2	$this->channel->close();
253		}
254	3	}
255		}
256

mmucklo / DtcQueueBundle

Push — master ( f30a66...783f1c )

JobManager::prioritySave() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like