Alchemy\Queue\Amqp\AmqpMessageQueue

Complexity

Total Complexity

8

Size/Duplication

Total Lines	94
Duplicated Lines	19.15 %

Coupling/Cohesion

Components	1
Dependencies	8

Test Coverage

Coverage

100%

Importance

Changes	3
Bugs	0	Features	2

6 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	setLogger()	0	4	1
A	publish()	0	11	2
A	ackMessage()	9	9	1
A	handle()	0	20	2
A	nackMessage()	9	9	1

<?php

namespace Alchemy\Queue\Amqp;

use Alchemy\Queue\Message;
use Alchemy\Queue\MessageHandlerResolver;
use Alchemy\Queue\MessageHandlingException;
use Alchemy\Queue\MessagePublishingException;
use Alchemy\Queue\MessageQueue;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use Ramsey\Uuid\Uuid;

class AmqpMessageQueue implements MessageQueue, LoggerAwareInterface
{
    /**
     * @var \AMQPExchange
     */
    private $exchange;

    /**
     * @var \AMQPQueue
     */
    private $queue;

    /**
     * @var LoggerInterface
     */
    private $logger;

    public function __construct(\AMQPExchange $exchange, \AMQPQueue $queue)
    {
        $this->exchange = $exchange;
        $this->queue = $queue;
        $this->logger = new NullLogger();
    }

    /**
     * Sets a logger instance on the object
     *
     * @param LoggerInterface $logger
     * @return null
     */
    public function setLogger(LoggerInterface $logger)
    {
        $this->logger = $logger;
    }

    /**
     * @param Message $message
     */
    public function publish(Message $message)
    {
        $this->logger->debug('Publishing message to queue: ' . $this->queue->getName());

        $attributes = ['correlation_id' => $message->getCorrelationId()];
        $result = $this->exchange->publish($message->getBody(), $this->queue->getName(), AMQP_DURABLE, $attributes);

        if (! $result) {
            throw new MessagePublishingException();
        }
    }

    /**
     * @param MessageHandlerResolver $resolver
     */
    public function handle(MessageHandlerResolver $resolver)
    {
        $this->logger->debug('Consuming messages from AMQP queue: ' . $this->queue->getName());

        $this->queue->consume(function (\AMQPEnvelope $envelope, \AMQPQueue $queue) use ($resolver) {
            $message = new Message($envelope->getBody(), $envelope->getCorrelationId());
            $handler = $resolver->resolveHandler($message);

            try {
                $this->logger->debug('Dispatching message to handler');
                $handler->handle($message);
                $this->ackMessage($queue, $envelope);
            } catch (MessageHandlingException $exception) {
                $this->logger->error('Caught exception while handling message: ' . $exception->getMessage());
                $this->nackMessage($queue, $envelope);
            }

            return false;
        }, AMQP_NOPARAM, Uuid::uuid4());
    }

    private function ackMessage(\AMQPQueue $queue, \AMQPEnvelope $envelope)

This method seems to be duplicated in your project.

    {
        $this->logger->debug('ACK message', [
            'correlation_id' => $envelope->getCorrelationId(),
            'body' => $envelope->getBody()
        ]);

        $queue->ack($envelope->getDeliveryTag(), AMQP_NOPARAM);
    }

    private function nackMessage(\AMQPQueue $queue, \AMQPEnvelope $envelope)

This method seems to be duplicated in your project.

    {
        $this->logger->debug('NACK message', [
            'correlation_id' => $envelope->getCorrelationId(),
            'body' => $envelope->getBody()
        ]);

        $queue->nack($envelope->getDeliveryTag(), AMQP_NOPARAM);
    }
}