Tomaj\Hermes\Driver\RabbitMqDriver

Complexity

Total Complexity

5

Size/Duplication

Total Lines	78
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	3

Test Coverage

Coverage

33.33%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	1
A	send()	0	6	1
A	setupPriorityQueue()	0	4	1
A	wait()	0	20	2

<?php
declare(strict_types=1);

namespace Tomaj\Hermes\Driver;

use Closure;
use Exception;
use Tomaj\Hermes\Dispatcher;
use Tomaj\Hermes\MessageInterface;
use Tomaj\Hermes\MessageSerializer;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Channel\AMQPChannel;

/**
 * Class RabbitMqDriver
 * @package Tomaj\Hermes\Driver
 *
 * @deprecated use LazyRabbitMqDriver instead
 */
class RabbitMqDriver implements DriverInterface
{
    use SerializerAwareTrait;

    /**
     * @var AMQPChannel
     */
    private $channel;

    /**
     * @var string
     */
    private $queue;

    /**
     * @var array
     */
    private $amqpMessageProperties = [];

    /**
     * Create new RabbitMqDriver with provided channel.
     *
     * You have to create connection to rabbit, and setup queue outside of this class.
     * Handling connection to rabbit is up to you and you have to manage it.
     *
     * @see examples/rabbitmq folder
     *
     * @param AMQPChannel   $channel
     * @param string        $queue
     * @param array         $amqpMessageProperties
     */
    public function __construct(AMQPChannel $channel, string $queue, array $amqpMessageProperties = [])
    {
        $this->channel = $channel;
        $this->queue = $queue;
        $this->amqpMessageProperties = $amqpMessageProperties;
        $this->serializer = new MessageSerializer();
    }

    /**
     * {@inheritdoc}
     */
    public function send(MessageInterface $message, int $priority = Dispatcher::PRIORITY_MEDIUM): bool
    {
        $rabbitMessage = new AMQPMessage($this->serializer->serialize($message), $this->amqpMessageProperties);
        $this->channel->basic_publish($rabbitMessage, '', $this->queue);
        return true;
    }

    public function setupPriorityQueue(string $name, int $priority): void
    {
        throw new \Exception("AmazonSQS is not supporting priority queues now");
    }

    /**
     * {@inheritdoc}
     */
    public function wait(Closure $callback, array $priorities): void
    {
        $this->channel->basic_consume(
            $this->queue,
            '',
            false,
            false,
            false,
            false,
            function ($rabbitMessage) use ($callback) {
                $message = $this->serializer->unserialize($rabbitMessage->body);
                $callback($message);
                $rabbitMessage->delivery_info['channel']->basic_ack($rabbitMessage->delivery_info['delivery_tag']);
            }
        );

        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }
}