Initx\Querabilis\Driver\AmqpQueue

Complexity

Total Complexity

8

Size/Duplication

Total Lines	91
Duplicated Lines	0 %

Test Coverage

Coverage

96.43%

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	add()	0	7	2
A	poll()	0	9	2
A	peek()	0	11	2
A	__construct()	0	13	1
A	offer()	0	8	1

<?php declare(strict_types=1);

namespace Initx\Querabilis\Driver;

use Initx\Querabilis\Envelope;
use Initx\Querabilis\Exception\IllegalStateException;
use Initx\Querabilis\Queue;
use JMS\Serializer\SerializerInterface;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Message\AMQPMessage;

final class AmqpQueue implements Queue
{
    use HasFallbackSerializer;
    use HasDefaultRemoveAndElement;

    /**
     * @var SerializerInterface
     */
    private $serializer;

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

    /**
     * @var AbstractConnection
     */
    private $connection;

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

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

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

    public function __construct(
        AbstractConnection $connection,
        string $queue,
        string $exchange = '',
        string $routingKey = '',
        ?SerializerInterface $serializer = null
    ) {
        $this->connection = $connection;
        $this->channel = $connection->channel();
        $this->queue = $queue;
        $this->exchange = $exchange;
        $this->routingKey = $routingKey;
        $this->serializer = $this->fallbackSerializer($serializer);
    }

    public function add(Envelope $envelope): bool
    {
        if (!$this->offer($envelope)) {
            throw new IllegalStateException("Could not write to AMQP");
        }

        return true;
    }

    public function offer(Envelope $envelope): bool
    {
        $serialized = $this->serializer->serialize($envelope, 'json');
        $message = new AMQPMessage($serialized);

        $this->channel->basic_publish($message, $this->exchange, $this->routingKey);

        return true;
    }

    public function poll(): ?Envelope
    {
        $message = $this->channel->basic_get($this->queue, true);

        if (!$message) {
            return null;
        }

        return $this->serializer->deserialize($message->body, Envelope::class, 'json');
    }

    public function peek(): ?Envelope
    {
        $message = $this->channel->basic_get($this->queue);

        if (!$message) {
            return null;
        }

        $this->channel->basic_nack($message->get('delivery_tag'), false, true);

        return $this->serializer->deserialize($message->body, Envelope::class, 'json');
    }
}