Queue::failed() - Code Metrics - mariano/disque-php - Measure and Improve Code Quality continuously with Scrutinizer

Queue::failed() A
last analyzed 2017-05-30 02:39 UTC

↳ Parent: Queue

Complexity

Conditions	1
Paths	1

Size

Total Lines	5
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

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

<?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			public function __construct(Client $client, $name)
52			{
53			$this->client = $client;
54			$this->name = $name;
55			$this->setMarshaler(new JobMarshaler());
56			}
57
58			/**
59			* Get the queue name
60			*
61			* @return string
62			*/
63			public function getName()
64			{
65			return $this->name;
66			}
67
68			/**
69			* Set Job marshaler
70			*
71			* @param MarshalerInterface Marshaler
72			* @return void
73			*/
74			public function setMarshaler(MarshalerInterface $marshaler)
75			{
76			$this->marshaler = $marshaler;
77			}
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			public function schedule(JobInterface $job, DateTime $when, array $options = [])
90			{
91			if (!isset($this->timeZone)) {
92			$this->timeZone = new DateTimeZone(self::DEFAULT_JOB_TIMEZONE);
93			}
94
95			$date = clone($when);
96			$date->setTimeZone($this->timeZone);
97			$now = new DateTime('now', $this->timeZone);
98			if ($date < $now) {
99			throw new InvalidArgumentException('Specified schedule time has passed');
100			}
101
102			$options['delay'] = ($date->getTimestamp() - $now->getTimestamp());
103			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			public function push(JobInterface $job, array $options = [])
114			{
115			$this->checkConnected();
116			$id = $this->client->addJob($this->name, $this->marshaler->marshal($job), $options);
117			$job->setId($id);
118			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			public function pull($timeout = 0)
130			{
131			$this->checkConnected();
132			$jobs = $this->client->getJob($this->name, [
133			'timeout' => $timeout,
134			'count' => 1,
135			'withcounters' => true
136			]);
137			if (empty($jobs)) {
138			return null;
139			}
140			$jobData = $jobs[0];
141			$job = $this->marshaler->unmarshal($jobData[Response::KEY_BODY]);
142
143			$job->setId($jobData[Response::KEY_ID]);
144			$job->setNacks($jobData[Counters::KEY_NACKS]);
145			$job->setAdditionalDeliveries(
146			$jobData[Counters::KEY_ADDITIONAL_DELIVERIES]
147			);
148
149			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			public function processing(JobInterface $job)
159			{
160			$this->checkConnected();
161			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			public function processed(JobInterface $job)
171			{
172			$this->checkConnected();
173			$this->client->ackJob($job->getId());
174			}
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			public function failed(JobInterface $job)
185			{
186			$this->checkConnected();
187			$this->client->nack($job->getId());
188			}
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			private function checkConnected()
197			{
198			if (!$this->client->isConnected()) {
199			$this->client->connect();
200			}
201			}
202			}
203

mariano / disque-php

Queue::failed() A last analyzed 2017-05-30 02:39 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Queue::failed() A
last analyzed 2017-05-30 02:39 UTC