Bernard\Driver\SqsDriver

Complexity

Total Complexity

28

Size/Duplication

Total Lines	226
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes

0

13 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	1
A	listQueues()	0	19	4
A	createQueue()	0	15	3
A	isFifoQueue()	0	4	1
A	endsWith()	0	9	2
A	countMessages()	0	15	2
A	pushMessage()	0	16	2
B	popMessage()	0	24	5
A	acknowledgeMessage()	0	9	1
A	peekQueue()	0	4	1
A	removeQueue()	0	8	1
A	info()	0	6	1
A	resolveUrl()	0	14	4

<?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) {
            $result = $this->sqs->createQueue([
                'QueueName' => $queueName,
                'Attributes' => [
                    'FifoQueue' => $this->isFifoQueue($queueName) ? 'true' : 'false',
                ],
            ]);

            $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)){

Expected 1 space after IF keyword; 0 found

            $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.');
    }
}