SqsDriver::queueExists() - Code Metrics - Inspection of "Amazon SQS automatic queue creation (#327)" - bernardphp/bernard - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1095c1...28e5b2 )

by Jeroen

created 2017-09-23 19:22 UTC

SqsDriver::queueExists() B

↳ Parent: SqsDriver

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;

use Aws\Sqs\Exception\SqsException;
use Aws\Sqs\SqsClient;

/**
 * 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
 */
class SqsDriver extends AbstractPrefetchDriver
{
    const AWS_SQS_FIFO_SUFFIX = '.fifo';
    const AWS_SQS_EXCEPTION_BAD_REQUEST = 400;

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

bernardphp / bernard

GitHub Access Token became invalid

Push — master ( 1095c1...28e5b2 )

SqsDriver::queueExists() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like