Tomaj\Hermes\Driver\RabbitMqDriver

Complexity

Total Complexity

4

Size/Duplication

Total Lines	65
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Test Coverage

Coverage

33.33%

Importance

Changes	10
Bugs	2	Features	4

3 Methods

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

<?php

namespace Tomaj\Hermes\Driver;

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

class RabbitMqDriver implements DriverInterface
{
    use SerializerAwareTrait;

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

    /**
     * @var string
     */
    private $queue;
    
    /**
     * 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
     */
    public function __construct(AMQPChannel $channel, $queue)
    {
        $this->channel = $channel;
        $this->queue = $queue;
        $this->serializer = new MessageSerializer();
    }

    /**
     * {@inheritdoc}
     */
    public function send(MessageInterface $message)
    {
        $rabbitMessage = new AMQPMessage($this->serializer->serialize($message));
        $this->channel->basic_publish($rabbitMessage, '', $this->queue);
    }

    /**
     * {@inheritdoc}
     */
    public function wait(Closure $callback)
    {
        $this->channel->basic_consume(
            $this->queue,
            '',
            false,
            true,
            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();
        }
    }
}