SqsDriver::isFifoQueue() - Code Metrics - Inspection of "AWS SQS FIFO Queue support (#318)" - bernardphp/bernard - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( c1e675...15e258 )

by Jeroen

created 2017-09-15 10:24 UTC

SqsDriver::isFifoQueue() A

↳ Parent: SqsDriver

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	4
rs	10
c	0
b	0
f	0
cc	1
eloc	2
nc	1
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';

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

bernardphp / bernard

GitHub Access Token became invalid

Push — master ( c1e675...15e258 )

SqsDriver::isFifoQueue() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like