Happyr\Mq2phpBundle\Service\ConsumerWrapper

Complexity

Total Complexity

10

Size/Duplication

Total Lines	118
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Test Coverage

Coverage

24.32%

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	11	1
A	consume()	0	14	3
A	setLogger()	0	4	1
A	log()	0	6	2
B	doConsume()	0	24	3

<?php

namespace Happyr\Mq2phpBundle\Service;

use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use SimpleBus\Asynchronous\Consumer\SerializedEnvelopeConsumer;

/**
 * This class delegates a message to the CommandConsumer or EventConsumer depending on the queue name.
 *
 * @author Tobias Nyholm <tobias.nyholm@gmail.com>
 */
class ConsumerWrapper implements LoggerAwareInterface
{
    /**
     * @var SerializedEnvelopeConsumer
     */
    private $commandConsumer;

    /**
     * @var SerializedEnvelopeConsumer
     */
    private $eventConsumer;

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

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

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

    /**
     * @param string                     $commandQueueName
     * @param string                     $eventQueueName
     * @param SerializedEnvelopeConsumer $commandConsumer
     * @param SerializedEnvelopeConsumer $eventConsumer
     */
    public function __construct(
        $commandQueueName,
        $eventQueueName,
        SerializedEnvelopeConsumer $commandConsumer = null,
        SerializedEnvelopeConsumer $eventConsumer = null
    ) {
        $this->commandConsumer = $commandConsumer;
        $this->eventConsumer = $eventConsumer;
        $this->commandQueueName = $commandQueueName;
        $this->eventQueueName = $eventQueueName;
    }

    /**
     * @param $queueName
     * @param $message
     */
    public function consume($queueName, $message)
    {
        $this->log('info', sprintf('Consuming data from queue: %s', $queueName));

        if ($queueName === $this->eventQueueName) {
            $this->doConsume($queueName, $message, $this->eventConsumer);
            $this->log('info', sprintf('Data from queue %s was consumed by the event consumer', $queueName));
        } elseif ($queueName === $this->commandQueueName) {
            $this->doConsume($queueName, $message, $this->commandConsumer);
            $this->log('info', sprintf('Data from queue %s was consumed by the command consumer', $queueName));
        } else {
            $this->doConsume($queueName, $message);
        }
    }

    /**
     * @param LoggerInterface $logger
     */
    public function setLogger(LoggerInterface $logger)
    {
        $this->logger = $logger;
    }

    /**
     * @param string $level
     * @param string $message
     * @param array  $context
     */
    private function log($level, $message, array $context = [])
    {
        if ($this->logger) {
            $this->logger->log($level, $message, $context);
        }
    }

    /**
     * Consume a message and make sure we log errors.
     *
     * @param string                     $queueName
     * @param mixed                      $message
     * @param SerializedEnvelopeConsumer $consumer
     *
     * @throws \Exception
     */
    private function doConsume($queueName, $message, SerializedEnvelopeConsumer $consumer = null)
    {
        if ($consumer === null) {
            $exceptionMessage = sprintf('No consumer was found for queue named "%s"', $queueName);
            $this->log('alert', $exceptionMessage, ['message' => $message]);

            throw new \RuntimeException($exceptionMessage);
        }

        try {
            $consumer->consume($message);
        } catch (\Exception $e) {
            $this->log(
                'error',
                sprintf('Tried to handle message from queue %s but failed', $queueName),
                [
                    'exception' => $e,
                    'message' => $message,
                ]
            );

            throw $e;
        }
    }
}