ShiftOneLabs\LaravelSqsFifoQueue\SqsFifoQueue

Complexity

Total Complexity

22

Size/Duplication

Total Lines	239
Duplicated Lines	0 %

Test Coverage

Coverage

98.46%

Importance

Changes	2
Bugs	1	Features	0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	getMeta()	0	5	1
A	getMetaPayload()	0	13	4
A	appendPayload()	0	13	3
A	pushRaw()	0	13	2
A	createPayload()	0	25	4
A	createPayloadArray()	0	5	1
A	setSqs()	0	5	1
A	later()	0	3	1
A	getDeduplicationId()	0	19	4

<?php

namespace ShiftOneLabs\LaravelSqsFifoQueue;

use LogicException;
use Aws\Sqs\SqsClient;
use BadMethodCallException;
use InvalidArgumentException;
use Illuminate\Queue\SqsQueue;
use ShiftOneLabs\LaravelSqsFifoQueue\Support\Arr;
use ShiftOneLabs\LaravelSqsFifoQueue\Contracts\Queue\Deduplicator;

class SqsFifoQueue extends SqsQueue
{
    /**
     * The message group id of the fifo pipe in the queue.
     *
     * @var string
     */
    protected $group;

    /**
     * The driver to generate the deduplication id for the message.
     *
     * @var string
     */
    protected $deduplicator;

    /**
     * Create a new Amazon SQS queue instance.
     *
     * @param  \Aws\Sqs\SqsClient  $sqs
     * @param  string  $default
     * @param  string  $prefix
     * @param  string  $group
     * @param  string  $deduplicator
     *
     * @return void
     */
    public function __construct(SqsClient $sqs, $default, $prefix = '', $group = '', $deduplicator = '')
    {
        parent::__construct($sqs, $default, $prefix);

        $this->group = $group;
        $this->deduplicator = $deduplicator;
    }

    /**
     * Set the underlying SQS instance.
     *
     * @param  \Aws\Sqs\SqsClient  $sqs
     *
     * @return \ShiftOneLabs\LaravelSqsFifoQueue\SqsFifoQueue
     */
    public function setSqs(SqsClient $sqs)
    {
        $this->sqs = $sqs;

        return $this;
    }

    /**
     * Push a raw payload onto the queue.
     *
     * @param  string  $payload
     * @param  string|null  $queue
     * @param  array  $options
     *
     * @return mixed
     */
    public function pushRaw($payload, $queue = null, array $options = [])
    {
        $message = [
            'QueueUrl' => $this->getQueue($queue), 'MessageBody' => $payload, 'MessageGroupId' => $this->getMeta($payload, 'group', $this->group),
        ];

        if (($deduplication = $this->getDeduplicationId($payload, $queue)) !== false) {
            $message['MessageDeduplicationId'] = $deduplication;
        }

        $response = $this->sqs->sendMessage($message);

        return $response->get('MessageId');
    }

    /**
     * Push a new job onto the queue after a delay.
     *
     * @param  \DateTime|int  $delay
     * @param  string  $job
     * @param  mixed  $data
     * @param  string|null  $queue
     *
     * @return mixed
     *
     * @throws BadMethodCallException
     */
    public function later($delay, $job, $data = '', $queue = null)
    {
        throw new BadMethodCallException('FIFO queues do not support per-message delays.');
    }

    /**
     * Get the deduplication id for the given driver.
     *
     * @param  string  $payload
     * @param  string  $queue
     *
     * @return string|bool
     *
     * @throws InvalidArgumentException
     */
    protected function getDeduplicationId($payload, $queue)
    {
        $driver = $this->getMeta($payload, 'deduplicator', $this->deduplicator);

        if (empty($driver)) {
            return false;
        }

        if ($this->container->bound($key = 'queue.sqs-fifo.deduplicator.'.$driver)) {
            $deduplicator = $this->container->make($key);

            if ($deduplicator instanceof Deduplicator) {
                return $deduplicator->generate($payload, $queue);
            }

            throw new InvalidArgumentException(sprintf('Deduplication method [%s] must resolve to a %s implementation.', $driver, Deduplicator::class));
        }

        throw new InvalidArgumentException(sprintf('Unsupported deduplication method [%s].', $driver));
    }

    /**
     * Create a payload string from the given job and data.
     *
     * @param  mixed  $job
     * @param  mixed  $data
     * @param  string|null  $queue
     *
     * @return string
     *
     * @throws \LogicException
     * @throws \InvalidArgumentException
     * @throws \Illuminate\Queue\InvalidPayloadException
     */
    protected function createPayload($job, $data = '', $queue = null)
    {
        $payload = parent::createPayload($job, $data, $queue);

        if (!is_object($job)) {
            return $payload;
        }

        // Laravel 5.4 reworked payload generate. If the parent class has
        // the `createPayloadArray` method, it has already been called
        // through the parent call to the "createPayload" method.
        if (method_exists(get_parent_class($this), 'createPayloadArray')) {
            return $payload;
        }

        // Laravel <= 5.3 has the `setMeta` method. This is the method
        // used to add meta data to the payload generated by the
        // parent call to `createPayload` above.
        if (method_exists($this, 'setMeta')) {
            return $this->appendPayload($payload, $job);
        }

        // If neither of the above methods exist, we must be on a version
        // of Laravel that is not currently supported.
        throw new LogicException('"createPayloadArray" and "setMeta" methods both missing. This version of Laravel not supported.');
    }

    /**
     * Append meta data to the payload for Laravel <= 5.3.
     *
     * @param  string  $payload
     * @param  mixed  $job
     *
     * @return string
     */
    protected function appendPayload($payload, $job)
    {
        $meta = $this->getMetaPayload($job);

        if (array_key_exists('group', $meta)) {
            $payload = $this->setMeta($payload, 'group', $meta['group']);

The method setMeta() does not exist on ShiftOneLabs\LaravelSqsFifoQueue\SqsFifoQueue.

        }

        if (array_key_exists('deduplicator', $meta)) {
            $payload = $this->setMeta($payload, 'deduplicator', $meta['deduplicator']);
        }

        return $payload;
    }

    /**
     * Create a payload array from the given job and data.
     *
     * @param  mixed  $job
     * @param  mixed  $data
     * @param  string|null  $queue
     *
     * @return array
     */
    protected function createPayloadArray($job, $data = '', $queue = null)
    {
        return array_merge(
            parent::createPayloadArray($job, $data, $queue),
            $this->getMetaPayload($job)
        );
    }

    /**
     * Get the meta data to add to the payload.
     *
     * @param  mixed  $job
     *
     * @return array
     */
    protected function getMetaPayload($job)
    {
        if (!is_object($job)) {
            return [];
        }

        return array_filter(
            [
                'group' => isset($job->messageGroupId) ? $job->messageGroupId : null,
                'deduplicator' => isset($job->deduplicator) ? $job->deduplicator : null,
            ],
            function ($value) {
                return $value !== null;
            }
        );
    }

    /**
     * Get additional meta from a payload string.
     *
     * @param  string  $payload
     * @param  string  $key
     * @param  mixed  $default
     *
     * @return mixed
     */
    protected function getMeta($payload, $key, $default = null)
    {
        $payload = json_decode($payload, true);

        return Arr::get($payload, $key, $default);
    }
}