Driver::queueExists() - Code Metrics - Inspection of "Relocate drivers (#337)" - bernardphp/bernard - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1144b4...828920 )

by Márk

created 2018-02-14 00:54 UTC

Driver::queueExists() B

↳ Parent: Driver

Complexity

Conditions	5
Paths	5

Size

Total Lines	17
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	17
rs	8.8571
c	0
b	0
f	0
cc	5
eloc	11
nc	5
nop	1

<?php

namespace Bernard\Driver\Sqs;

use Aws\Sqs\Exception\SqsException;
use Aws\Sqs\SqsClient;
use Bernard\Driver\AbstractPrefetchDriver;

/**
 * Implements a Driver for use with AWS SQS client API:
 * @link http://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html
 *
 * @package Bernard
 */
final class Driver extends AbstractPrefetchDriver
{
    const AWS_SQS_FIFO_SUFFIX = '.fifo';
    const AWS_SQS_EXCEPTION_BAD_REQUEST = 400;

    private $sqs;
    private $queueUrls;

    /**
     * @param SqsClient $sqs
     * @param array     $queueUrls
     * @param int|null  $prefetch
     */
    public function __construct(SqsClient $sqs, array $queueUrls = [], $prefetch = null)
    {
        parent::__construct($prefetch);

        $this->sqs = $sqs;
        $this->queueUrls = $queueUrls;
    }

    /**
     * {@inheritdoc}
     */
    public function listQueues()
    {
        $result = $this->sqs->listQueues();

        if (!$queueUrls = $result->get('QueueUrls')) {
            return array_keys($this->queueUrls);
        }

        foreach ($queueUrls as $queueUrl) {
            if (in_array($queueUrl, $this->queueUrls)) {
                continue;
            }

            $queueName = current(array_reverse(explode('/', $queueUrl)));
            $this->queueUrls[$queueName] = $queueUrl;
        }

        return array_keys($this->queueUrls);
    }

    /**
     * {@inheritdoc}
     *
     * @link http://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html#createqueue
     *
     * @return void
     *
     * @throws SqsException
     */
    public function createQueue($queueName)
    {
        if($this->queueExists($queueName)){

            return;
        }

        $parameters = [
            'QueueName' => $queueName,
        ];

        if($this->isFifoQueue($queueName)) {

            $parameters['Attributes'] = [
                'FifoQueue' => 'true',
            ];
        }

        $result = $this->sqs->createQueue($parameters);

        $this->queueUrls[$queueName] = $result['QueueUrl'];
    }

    /**
     * @param string $queueName
     *
     * @return bool
     *
     * @throws SqsException
     */
    private function queueExists($queueName)
    {
        try {
            $this->resolveUrl($queueName);
            return true;
        } catch (\InvalidArgumentException $exception) {
            return false;
        } catch (SqsException $exception) {
            if ($previousException = $exception->getPrevious()) {
                if ($previousException->getCode() === self::AWS_SQS_EXCEPTION_BAD_REQUEST) {
                    return false;
                }
            }

            throw $exception;
        }
    }

    /**
     * @param string $queueName
     *
     * @return bool
     */
    private function isFifoQueue($queueName)
    {
        return $this->endsWith($queueName, self::AWS_SQS_FIFO_SUFFIX);
    }

    /**
     * @param string $haystack
     * @param string $needle
     *
     * @return bool
     */
    private function endsWith($haystack, $needle)
    {
        $length = strlen($needle);
        if ($length === 0) {
            return true;
        }

        return (substr($haystack, -$length) === $needle);
    }

    /**
     * {@inheritdoc}
     */
    public function countMessages($queueName)
    {
        $queueUrl = $this->resolveUrl($queueName);

        $result = $this->sqs->getQueueAttributes([
            'QueueUrl' => $queueUrl,
            'AttributeNames' => ['ApproximateNumberOfMessages'],
        ]);

        if (isset($result['Attributes']['ApproximateNumberOfMessages'])) {
            return (int)$result['Attributes']['ApproximateNumberOfMessages'];
        }

        return 0;
    }

    /**
     * {@inheritdoc}
     */
    public function pushMessage($queueName, $message)
    {
        $queueUrl = $this->resolveUrl($queueName);

        $parameters = [
            'QueueUrl' => $queueUrl,
            'MessageBody' => $message,
        ];

        if($this->isFifoQueue($queueName)){

            $parameters['MessageGroupId'] = __METHOD__;
            $parameters['MessageDeduplicationId'] = md5($message);
        }

        $this->sqs->sendMessage($parameters);
    }

    /**
     * {@inheritdoc}
     */
    public function popMessage($queueName, $duration = 5)
    {
        if ($message = $this->cache->pop($queueName)) {
            return $message;
        }

        $queueUrl = $this->resolveUrl($queueName);

        $result = $this->sqs->receiveMessage([
            'QueueUrl' => $queueUrl,
            'MaxNumberOfMessages' => $this->prefetch,
            'WaitTimeSeconds' => $duration,
        ]);

        if (!$result || !$messages = $result->get('Messages')) {
            return [null, null];
        }

        foreach ($messages as $message) {
            $this->cache->push($queueName, [$message['Body'], $message['ReceiptHandle']]);
        }

        return $this->cache->pop($queueName);
    }

    /**
     * {@inheritdoc}
     */
    public function acknowledgeMessage($queueName, $receipt)
    {
        $queueUrl = $this->resolveUrl($queueName);

        $this->sqs->deleteMessage([
            'QueueUrl' => $queueUrl,
            'ReceiptHandle' => $receipt,
        ]);
    }

    /**
     * {@inheritdoc}
     */
    public function peekQueue($queueName, $index = 0, $limit = 20)
    {
        return [];
    }

    /**
     * {@inheritdoc}
     *
     * @link http://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html#deletequeue
     */
    public function removeQueue($queueName)
    {
        $queueUrl = $this->resolveUrl($queueName);

        $this->sqs->deleteQueue([
            'QueueUrl' => $queueUrl,
        ]);
    }

    /**
     * {@inheritdoc}
     */
    public function info()
    {
        return [
            'prefetch' => $this->prefetch,
        ];
    }

    /**
     * AWS works with queue URLs rather than queue names. Returns either queue URL (if queue exists) for given name or null if not.
     *
     * @param string $queueName
     *
     * @return mixed
     *
     * @throws SqsException
     */
    private function resolveUrl($queueName)
    {
        if (isset($this->queueUrls[$queueName])) {
            return $this->queueUrls[$queueName];
        }

        $result = $this->sqs->getQueueUrl(['QueueName' => $queueName]);

        if ($result && $queueUrl = $result->get('QueueUrl')) {
            return $this->queueUrls[$queueName] = $queueUrl;
        }

        throw new \InvalidArgumentException('Queue "' . $queueName .'" cannot be resolved to an url.');
    }
}


1			<?php
2
3			namespace Bernard\Driver\Sqs;
4
5			use Aws\Sqs\Exception\SqsException;
6			use Aws\Sqs\SqsClient;
7			use Bernard\Driver\AbstractPrefetchDriver;
8
9			/**
10			* Implements a Driver for use with AWS SQS client API:
11			* @link http://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html
12			*
13			* @package Bernard
14			*/
15			final class Driver extends AbstractPrefetchDriver
16			{
17			const AWS_SQS_FIFO_SUFFIX = '.fifo';
18			const AWS_SQS_EXCEPTION_BAD_REQUEST = 400;
19
20			private $sqs;
21			private $queueUrls;
22
23			/**
24			* @param SqsClient $sqs
25			* @param array $queueUrls
26			* @param int\|null $prefetch
27			*/
28			public function __construct(SqsClient $sqs, array $queueUrls = [], $prefetch = null)
29			{
30			parent::__construct($prefetch);
31
32			$this->sqs = $sqs;
33			$this->queueUrls = $queueUrls;
34			}
35
36			/**
37			* {@inheritdoc}
38			*/
39			public function listQueues()
40			{
41			$result = $this->sqs->listQueues();
42
43			if (!$queueUrls = $result->get('QueueUrls')) {
44			return array_keys($this->queueUrls);
45			}
46
47			foreach ($queueUrls as $queueUrl) {
48			if (in_array($queueUrl, $this->queueUrls)) {
49			continue;
50			}
51
52			$queueName = current(array_reverse(explode('/', $queueUrl)));
53			$this->queueUrls[$queueName] = $queueUrl;
54			}
55
56			return array_keys($this->queueUrls);
57			}
58
59			/**
60			* {@inheritdoc}
61			*
62			* @link http://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html#createqueue
63			*
64			* @return void
65			*
66			* @throws SqsException
67			*/
68			public function createQueue($queueName)
69			{
70			if($this->queueExists($queueName)){
			0 ignored issues – show Coding Style introduced 2017-09-15 12:19 UTC by Report Bug Copy Issue Report Expected 1 space after IF keyword; 0 found Loading history...
71			return;
72			}
73
74			$parameters = [
75			'QueueName' => $queueName,
76			];
77
78			if($this->isFifoQueue($queueName)) {
			0 ignored issues – show Coding Style introduced 2017-07-17 21:30 UTC by Report Bug Copy Issue Report Expected 1 space after IF keyword; 0 found Loading history...
79			$parameters['Attributes'] = [
80			'FifoQueue' => 'true',
81			];
82			}
83
84			$result = $this->sqs->createQueue($parameters);
85
86			$this->queueUrls[$queueName] = $result['QueueUrl'];
87			}
88
89			/**
90			* @param string $queueName
91			*
92			* @return bool
93			*
94			* @throws SqsException
95			*/
96			private function queueExists($queueName)
97			{
98			try {
99			$this->resolveUrl($queueName);
100			return true;
101			} catch (\InvalidArgumentException $exception) {
102			return false;
103			} catch (SqsException $exception) {
104			if ($previousException = $exception->getPrevious()) {
105			if ($previousException->getCode() === self::AWS_SQS_EXCEPTION_BAD_REQUEST) {
106			return false;
107			}
108			}
109
110			throw $exception;
111			}
112			}
113
114			/**
115			* @param string $queueName
116			*
117			* @return bool
118			*/
119			private function isFifoQueue($queueName)
120			{
121			return $this->endsWith($queueName, self::AWS_SQS_FIFO_SUFFIX);
122			}
123
124			/**
125			* @param string $haystack
126			* @param string $needle
127			*
128			* @return bool
129			*/
130			private function endsWith($haystack, $needle)
131			{
132			$length = strlen($needle);
133			if ($length === 0) {
134			return true;
135			}
136
137			return (substr($haystack, -$length) === $needle);
138			}
139
140			/**
141			* {@inheritdoc}
142			*/
143			public function countMessages($queueName)
144			{
145			$queueUrl = $this->resolveUrl($queueName);
146
147			$result = $this->sqs->getQueueAttributes([
148			'QueueUrl' => $queueUrl,
149			'AttributeNames' => ['ApproximateNumberOfMessages'],
150			]);
151
152			if (isset($result['Attributes']['ApproximateNumberOfMessages'])) {
153			return (int)$result['Attributes']['ApproximateNumberOfMessages'];
154			}
155
156			return 0;
157			}
158
159			/**
160			* {@inheritdoc}
161			*/
162			public function pushMessage($queueName, $message)
163			{
164			$queueUrl = $this->resolveUrl($queueName);
165
166			$parameters = [
167			'QueueUrl' => $queueUrl,
168			'MessageBody' => $message,
169			];
170
171			if($this->isFifoQueue($queueName)){
			0 ignored issues – show Coding Style introduced 2017-07-18 08:35 UTC by Report Bug Copy Issue Report Expected 1 space after IF keyword; 0 found Loading history...
172			$parameters['MessageGroupId'] = __METHOD__;
173			$parameters['MessageDeduplicationId'] = md5($message);
174			}
175
176			$this->sqs->sendMessage($parameters);
177			}
178
179			/**
180			* {@inheritdoc}
181			*/
182			public function popMessage($queueName, $duration = 5)
183			{
184			if ($message = $this->cache->pop($queueName)) {
185			return $message;
186			}
187
188			$queueUrl = $this->resolveUrl($queueName);
189
190			$result = $this->sqs->receiveMessage([
191			'QueueUrl' => $queueUrl,
192			'MaxNumberOfMessages' => $this->prefetch,
193			'WaitTimeSeconds' => $duration,
194			]);
195
196			if (!$result \|\| !$messages = $result->get('Messages')) {
197			return [null, null];
198			}
199
200			foreach ($messages as $message) {
201			$this->cache->push($queueName, [$message['Body'], $message['ReceiptHandle']]);
202			}
203
204			return $this->cache->pop($queueName);
205			}
206
207			/**
208			* {@inheritdoc}
209			*/
210			public function acknowledgeMessage($queueName, $receipt)
211			{
212			$queueUrl = $this->resolveUrl($queueName);
213
214			$this->sqs->deleteMessage([
215			'QueueUrl' => $queueUrl,
216			'ReceiptHandle' => $receipt,
217			]);
218			}
219
220			/**
221			* {@inheritdoc}
222			*/
223			public function peekQueue($queueName, $index = 0, $limit = 20)
224			{
225			return [];
226			}
227
228			/**
229			* {@inheritdoc}
230			*
231			* @link http://docs.aws.amazon.com/aws-sdk-php/v3/api/api-sqs-2012-11-05.html#deletequeue
232			*/
233			public function removeQueue($queueName)
234			{
235			$queueUrl = $this->resolveUrl($queueName);
236
237			$this->sqs->deleteQueue([
238			'QueueUrl' => $queueUrl,
239			]);
240			}
241
242			/**
243			* {@inheritdoc}
244			*/
245			public function info()
246			{
247			return [
248			'prefetch' => $this->prefetch,
249			];
250			}
251
252			/**
253			* AWS works with queue URLs rather than queue names. Returns either queue URL (if queue exists) for given name or null if not.
254			*
255			* @param string $queueName
256			*
257			* @return mixed
258			*
259			* @throws SqsException
260			*/
261			private function resolveUrl($queueName)
262			{
263			if (isset($this->queueUrls[$queueName])) {
264			return $this->queueUrls[$queueName];
265			}
266
267			$result = $this->sqs->getQueueUrl(['QueueName' => $queueName]);
268
269			if ($result && $queueUrl = $result->get('QueueUrl')) {
270			return $this->queueUrls[$queueName] = $queueUrl;
271			}
272
273			throw new \InvalidArgumentException('Queue "' . $queueName .'" cannot be resolved to an url.');
274			}
275			}
276

bernardphp / bernard

GitHub Access Token became invalid

Push — master ( 1144b4...828920 )

Driver::queueExists() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like