Da\Mailer\Queue\Backend\RabbitMq\RabbitMqQueueStoreAdapter

Complexity

Total Complexity

13

Size/Duplication

Total Lines	134
Duplicated Lines	0 %

Importance

Changes	2
Bugs	0	Features	0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	getConnection()	0	3	1
A	enqueue()	0	12	2
A	isEmpty()	0	7	2
A	ack()	0	10	2
A	__construct()	0	7	1
A	init()	0	6	1
A	createPayload()	0	7	2
A	dequeue()	0	19	2

<?php

namespace Da\Mailer\Queue\Backend\RabbitMq;

use Da\Mailer\Queue\Backend\MailJobInterface;
use Da\Mailer\Queue\Backend\QueueStoreAdapterInterface;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use phpseclib3\Crypt\Random;

class RabbitMqQueueStoreAdapter implements QueueStoreAdapterInterface
{
    /**
     * @var string
     */
    private $queueName;

    /**
     * @var int
     */
    private $expireTime;

    /**
     * @var RabbitMqQueueConnection
     */
    protected $connection;

    /**
     * @param RabbitMqQueueConnection $connection
     * @param $queueName
     * @param $expireTime
     */
    public function __construct(RabbitMqQueueConnection $connection, $queueName = 'mail_queue', $expireTime = 60)
    {
        $this->connection = $connection;
        $this->expireTime = $expireTime;
        $this->queueName = $queueName;

        $this->init();
    }

    /**
     * @return
     */
    public function init()
    {
        $this->getConnection()
            ->connect();

        return $this;
    }

    /**
     * @inheritDoc
     */
    public function getConnection()
    {
        return $this->connection;
    }

    /**
     * @var RabbitMqJob|MailJobInterface $mailJob
     *
     * @return bool;
     */
    public function enqueue(MailJobInterface $mailJob)
    {
        try {
            /** @var AMQPChannel $chanel */
            $chanel = $this->getConnection()->getInstance();
            $chanel->queue_declare($this->queueName, false, false, false, false);
            $message = new AMQPMessage($this->createPayload($mailJob));
            $chanel->basic_publish($message, '', $this->queueName);

            return true;
        } catch (\Exception $exception) {
            return false;
        }
    }

    /**
     * @inheritDoc
     * @return RabbitMqJob|MailJobInterface|null
     */
    public function dequeue()
    {
        if ($this->isEmpty()) {
            return null;
        }

        /** @var AMQPChannel $chanel */
        $chanel = $this->getConnection()->getInstance();

        /** @var AMQPMessage $message */
        $message = $chanel->basic_get($this->queueName);

        $data = json_decode($message->body, true);

        return new RabbitMqJob([
            'id' => $data['id'],
            'message' => $data['message'],
            'attempt' => $data['attempt'],
            'deliveryTag' => $message->delivery_info['delivery_tag'],

The property PhpAmqpLib\Message\AMQPMessage::$delivery_info has been deprecated.

        ]);
    }

    /**
     * @param RabbitMqJob $mailJob
     */
    public function ack(MailJobInterface $mailJob)
    {
        /** @var AMQPChannel $chanel */
        $chanel = $this->getConnection()->getInstance();
        if ($mailJob->isCompleted()) {
            $chanel->basic_ack($mailJob->getDeliveryTag(), false);

The method getDeliveryTag() does not exist on Da\Mailer\Queue\Backend\MailJobInterface. It seems like you code against a sub-type of Da\Mailer\Queue\Backend\MailJobInterface such as Da\Mailer\Queue\Backend\RabbitMq\RabbitMqJob.

            return;
        }

        $chanel->basic_nack($mailJob->getDeliveryTag(), false, true);
    }

    /**
     * @inheritDoc
     */
    public function isEmpty()
    {
        /** @var AMQPChannel $chanel */
        $chanel = $this->getConnection()->getInstance();
        $queueProperties = $chanel->queue_declare($this->queueName, false, false, false, false);

        return is_array($queueProperties) && $queueProperties[1] === 0;
    }

    /**
     * @param MailJobInterface $mailJob
     * @return false|string
     */
    protected function createPayload(MailJobInterface $mailJob)
    {
        return json_encode([
            'id' => $mailJob->isNewRecord() ? sha1(Random::string(32)) : $mailJob->getId(),

The method getId() does not exist on Da\Mailer\Queue\Backend\MailJobInterface. Since it exists in all sub-types, consider adding an abstract or default implementation to Da\Mailer\Queue\Backend\MailJobInterface.

            'attempt' => $mailJob->getAttempt(),

The method getAttempt() does not exist on Da\Mailer\Queue\Backend\MailJobInterface. Since it exists in all sub-types, consider adding an abstract or default implementation to Da\Mailer\Queue\Backend\MailJobInterface.

            'message' => $mailJob->getMessage(),
            'delivery_tag' => null,
        ]);
    }
}