Queue::push() - Code Metrics - Inspection of "Add Queue::getName (#42)" - mariano/disque-php - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 3f6cab...790d9e )

by Mariano

created 2017-03-08 18:57 UTC

Queue::push() A

↳ Parent: Queue

Complexity

Conditions	1
Paths	1

Size

Total Lines	7
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	1

Importance

Changes

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

<?php
namespace Disque\Queue;

use DateTime;
use DateTimeZone;
use Disque\Client;
use Disque\Queue\Marshal\JobMarshaler;
use Disque\Queue\Marshal\MarshalerInterface;
use InvalidArgumentException;
use Disque\Command\Response\JobsResponse AS Response;
use Disque\Command\Response\JobsWithCountersResponse AS Counters;

class Queue
{
    const DEFAULT_JOB_TIMEZONE = 'UTC';

    /**
     * Client
     *
     * @var Client
     */
    protected $client;

    /**
     * Name
     *
     * @var string
     */
    protected $name;

    /**
     * Job marshaler
     *
     * @var MarshalerInterface
     */
    private $marshaler;

    /**
     * Default time zone
     *
     * @var DateTimeZone
     */
    private $timeZone;

    /**
     * Create a queue
     *
     * @param Client $client Client
     * @param string $name Queue name
     */
    public function __construct(Client $client, $name)
    {
        $this->client = $client;
        $this->name = $name;
        $this->setMarshaler(new JobMarshaler());
    }
    
    /**
     * Get the queue name
     *
     * @return string
     */
    public function getName()
    {
        return $this->name;
    }

    /**
     * Set Job marshaler
     *
     * @param MarshalerInterface Marshaler
     * @return void
     */
    public function setMarshaler(MarshalerInterface $marshaler)
    {
        $this->marshaler = $marshaler;
    }

    /**
     * Pushes a job into the queue, setting it to be up for processing only at
     * the specific date & time.
     *
     * @param JobInterface $job Job
     * @param DateTime $when Date & time on when job should be ready for processing
     * @param array $options ADDJOB options sent to the client
     * @return JobInterface Job pushed
     * @throws InvalidArgumentException
     */
    public function schedule(JobInterface $job, DateTime $when, array $options = [])
    {
        if (!isset($this->timeZone)) {
            $this->timeZone = new DateTimeZone(self::DEFAULT_JOB_TIMEZONE);
        }

        $date = clone($when);
        $date->setTimeZone($this->timeZone);
        $now = new DateTime('now', $this->timeZone);
        if ($date < $now) {
            throw new InvalidArgumentException('Specified schedule time has passed');
        }

        $options['delay'] = ($date->getTimestamp() - $now->getTimestamp());
        return $this->push($job, $options);
    }

    /**
     * Pushes a job into the queue
     *
     * @param JobInterface $job Job
     * @param array $options ADDJOB options sent to the client
     * @return JobInterface Job pushed
     */
    public function push(JobInterface $job, array $options = [])
    {
        $this->checkConnected();
        $id = $this->client->addJob($this->name, $this->marshaler->marshal($job), $options);
        $job->setId($id);
        return $job;
    }

    /**
     * Pulls a single job from the queue (if none available, and if $timeout
     * specified, then wait only this much time for a job, otherwise return
     * `null`)
     *
     * @param int $timeout If specified, wait these many seconds
     * @return Job|null A job, or null if no job was found before timeout
     */
    public function pull($timeout = 0)
    {
        $this->checkConnected();
        $jobs = $this->client->getJob($this->name, [
            'timeout' => $timeout,
            'count' => 1,
            'withcounters' => true
        ]);
        if (empty($jobs)) {
            return null;
        }
        $jobData = $jobs[0];
        $job = $this->marshaler->unmarshal($jobData[Response::KEY_BODY]);

        $job->setId($jobData[Response::KEY_ID]);
        $job->setNacks($jobData[Counters::KEY_NACKS]);
        $job->setAdditionalDeliveries(
            $jobData[Counters::KEY_ADDITIONAL_DELIVERIES]
        );

        return $job;
    }

    /**
     * Marks that a Job is still being processed
     *
     * @param JobInterface $job Job
     * @return int Number of seconds that the job visibility was postponed
     */
    public function processing(JobInterface $job)
    {
        $this->checkConnected();
        return $this->client->working($job->getId());
    }

    /**
     * Acknowledges a Job as properly handled
     *
     * @param JobInterface $job Job
     * @return void
     */
    public function processed(JobInterface $job)
    {
        $this->checkConnected();
        $this->client->ackJob($job->getId());
    }

    /**
     * Marks the job as failed and returns it to the queue
     *
     * This increases the NACK counter of the job
     *
     * @param JobInterface $job
     * @return void
     */
    public function failed(JobInterface $job)
    {
        $this->checkConnected();
        $this->client->nack($job->getId());
    }

    /**
     * Check that we are connected to a node, and if not connect
     *
     * @return void
     * @throws Disque\Connection\ConnectionException
     */
    private function checkConnected()
    {
        if (!$this->client->isConnected()) {
            $this->client->connect();
        }
    }
}


1		<?php
2		namespace Disque\Queue;
3
4		use DateTime;
5		use DateTimeZone;
6		use Disque\Client;
7		use Disque\Queue\Marshal\JobMarshaler;
8		use Disque\Queue\Marshal\MarshalerInterface;
9		use InvalidArgumentException;
10		use Disque\Command\Response\JobsResponse AS Response;
11		use Disque\Command\Response\JobsWithCountersResponse AS Counters;
12
13		class Queue
14		{
15		const DEFAULT_JOB_TIMEZONE = 'UTC';
16
17		/**
18		* Client
19		*
20		* @var Client
21		*/
22		protected $client;
23
24		/**
25		* Name
26		*
27		* @var string
28		*/
29		protected $name;
30
31		/**
32		* Job marshaler
33		*
34		* @var MarshalerInterface
35		*/
36		private $marshaler;
37
38		/**
39		* Default time zone
40		*
41		* @var DateTimeZone
42		*/
43		private $timeZone;
44
45		/**
46		* Create a queue
47		*
48		* @param Client $client Client
49		* @param string $name Queue name
50		*/
51	27	public function __construct(Client $client, $name)
52		{
53	27	$this->client = $client;
54	27	$this->name = $name;
55	27	$this->setMarshaler(new JobMarshaler());
56	27	}
57
58		/**
59		* Get the queue name
60		*
61		* @return string
62		*/
63	1	public function getName()
64		{
65	1	return $this->name;
66		}
67
68		/**
69		* Set Job marshaler
70		*
71		* @param MarshalerInterface Marshaler
72		* @return void
73		*/
74	27	public function setMarshaler(MarshalerInterface $marshaler)
75		{
76	27	$this->marshaler = $marshaler;
77	27	}
78
79		/**
80		* Pushes a job into the queue, setting it to be up for processing only at
81		* the specific date & time.
82		*
83		* @param JobInterface $job Job
84		* @param DateTime $when Date & time on when job should be ready for processing
85		* @param array $options ADDJOB options sent to the client
86		* @return JobInterface Job pushed
87		* @throws InvalidArgumentException
88		*/
89	5	public function schedule(JobInterface $job, DateTime $when, array $options = [])
90		{
91	5	if (!isset($this->timeZone)) {
92	5	$this->timeZone = new DateTimeZone(self::DEFAULT_JOB_TIMEZONE);
93		}
94
95	5	$date = clone($when);
96	5	$date->setTimeZone($this->timeZone);
97	5	$now = new DateTime('now', $this->timeZone);
98	5	if ($date < $now) {
99	1	throw new InvalidArgumentException('Specified schedule time has passed');
100		}
101
102	4	$options['delay'] = ($date->getTimestamp() - $now->getTimestamp());
103	4	return $this->push($job, $options);
104		}
105
106		/**
107		* Pushes a job into the queue
108		*
109		* @param JobInterface $job Job
110		* @param array $options ADDJOB options sent to the client
111		* @return JobInterface Job pushed
112		*/
113	4	public function push(JobInterface $job, array $options = [])
114		{
115	4	$this->checkConnected();
116	4	$id = $this->client->addJob($this->name, $this->marshaler->marshal($job), $options);
117	4	$job->setId($id);
118	4	return $job;
119		}
120
121		/**
122		* Pulls a single job from the queue (if none available, and if $timeout
123		* specified, then wait only this much time for a job, otherwise return
124		* `null`)
125		*
126		* @param int $timeout If specified, wait these many seconds
127		* @return Job\|null A job, or null if no job was found before timeout
128		*/
129	6	public function pull($timeout = 0)
130		{
131	6	$this->checkConnected();
132	6	$jobs = $this->client->getJob($this->name, [
133	6	'timeout' => $timeout,
134	6	'count' => 1,
135		'withcounters' => true
136		]);
137	6	if (empty($jobs)) {
138	2	return null;
139		}
140	4	$jobData = $jobs[0];
141	4	$job = $this->marshaler->unmarshal($jobData[Response::KEY_BODY]);
142
143	4	$job->setId($jobData[Response::KEY_ID]);
144	4	$job->setNacks($jobData[Counters::KEY_NACKS]);
145	4	$job->setAdditionalDeliveries(
146	4	$jobData[Counters::KEY_ADDITIONAL_DELIVERIES]
147		);
148
149	4	return $job;
150		}
151
152		/**
153		* Marks that a Job is still being processed
154		*
155		* @param JobInterface $job Job
156		* @return int Number of seconds that the job visibility was postponed
157		*/
158	2	public function processing(JobInterface $job)
159		{
160	2	$this->checkConnected();
161	2	return $this->client->working($job->getId());
162		}
163
164		/**
165		* Acknowledges a Job as properly handled
166		*
167		* @param JobInterface $job Job
168		* @return void
169		*/
170	2	public function processed(JobInterface $job)
171		{
172	2	$this->checkConnected();
173	2	$this->client->ackJob($job->getId());
174	2	}
175
176		/**
177		* Marks the job as failed and returns it to the queue
178		*
179		* This increases the NACK counter of the job
180		*
181		* @param JobInterface $job
182		* @return void
183		*/
184	2	public function failed(JobInterface $job)
185		{
186	2	$this->checkConnected();
187	2	$this->client->nack($job->getId());
188	2	}
189
190		/**
191		* Check that we are connected to a node, and if not connect
192		*
193		* @return void
194		* @throws Disque\Connection\ConnectionException
195		*/
196	16	private function checkConnected()
197		{
198	16	if (!$this->client->isConnected()) {
199	5	$this->client->connect();
200		}
201	16	}
202		}
203

mariano / disque-php

Push — master ( 3f6cab...790d9e )

Queue::push() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like