CultuurNet\UDB3\EventHandling\AMQPPublisher

Complexity

Total Complexity

16

Size/Duplication

Total Lines	174
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

10 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	10	1
A	handle()	0	11	2
A	publishWithAMQP()	0	14	1
A	createAMQPMessage()	0	7	1
A	createAMQPBody()	0	14	2
A	createAMQPProperties()	0	9	1
A	getCorrelationId()	0	4	1
A	getContentType()	0	13	2
A	withContentType()	0	6	1
A	setContentType()	0	21	4

<?php
/**
 * @file
 */

namespace CultuurNet\UDB3\EventHandling;

use Broadway\Domain\DomainMessage;
use Broadway\EventHandling\EventListenerInterface;
use Broadway\Serializer\SerializableInterface;
use CultuurNet\UDB3\EventHandling\DomainMessage\SpecificationInterface;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\NullLogger;

class AMQPPublisher implements EventListenerInterface, LoggerAwareInterface
{
    use LoggerAwareTrait;

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

    /**
     * @var SpecificationInterface
     */
    private $domainMessageSpecification;

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

    /**
     * @var string[]
     */
    private $payloadClassToContentTypeMap;

    /**
     * @param $exchange
     * @param SpecificationInterface $domainMessageSpecification
     */
    public function __construct(
        AMQPChannel $channel,
        $exchange,
        SpecificationInterface $domainMessageSpecification
    ) {
        $this->channel = $channel;
        $this->exchange = $exchange;
        $this->domainMessageSpecification = $domainMessageSpecification;
        $this->logger = new NullLogger();
    }

    /**
     * @inheritdoc
     */
    public function handle(DomainMessage $domainMessage)
    {
        if ($this->domainMessageSpecification->isSatisfiedBy($domainMessage)) {
            $this->publishWithAMQP($domainMessage);
            return;
        }

        $this->logger->info(
            'specification ' . get_class($this->domainMessageSpecification) . ' was not satisified by message'
        );
    }

    /**
     * @param DomainMessage $domainMessage
     */
    private function publishWithAMQP(DomainMessage $domainMessage)
    {
        $key = null;

        $message = $this->createAMQPMessage($domainMessage);

        $this->logger->info('publishing to exchange ' . $this->exchange);

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

    /**
     * @param DomainMessage $domainMessage
     * @return AMQPMessage
     */
    private function createAMQPMessage(DomainMessage $domainMessage)
    {
        $body = $this->createAMQPBody($domainMessage);
        $properties = $this->createAMQPProperties($domainMessage);

        return new AMQPMessage($body, $properties);
    }

    /**
     * @param DomainMessage $domainMessage
     * @return string
     */
    private function createAMQPBody(DomainMessage $domainMessage)
    {
        $payload = $domainMessage->getPayload();

        if ($payload instanceof SerializableInterface) {
            return json_encode(
                $payload->serialize()
            );
        }

        throw new \RuntimeException(
            'Unable to serialize ' . get_class($payload)
        );
    }

    /**
     * @param DomainMessage $domainMessage
     * @return array
     */
    private function createAMQPProperties(DomainMessage $domainMessage)
    {
        $properties = [];

        $properties['content_type'] = $this->getContentType($domainMessage);
        $properties['correlation_id'] = $this->getCorrelationId($domainMessage);

        return $properties;
    }

    private function getCorrelationId(DomainMessage $domainMessage)
    {
        return $domainMessage->getId() . '-' . $domainMessage->getPlayhead();
    }

    private function getContentType(DomainMessage $domainMessage)
    {
        $payload = $domainMessage->getPayload();
        $payloadClass = get_class($payload);

        if (isset($this->payloadClassToContentTypeMap[$payloadClass])) {
            return $this->payloadClassToContentTypeMap[$payloadClass];
        }

        throw new \RuntimeException(
            'Unable to find the content type of ' . $payloadClass
        );
    }

    /**
     * @param string $payloadClass
     * @param string $contentType
     * @return static
     */
    public function withContentType($payloadClass, $contentType)
    {
        $c = clone $this;
        $c->setContentType($payloadClass, $contentType);
        return $c;
    }

    /**
     * @param string $payloadClass
     * @param string $contentType
     */
    private function setContentType($payloadClass, $contentType)
    {
        if (!is_string($payloadClass)) {
            throw new \InvalidArgumentException(
                'Value for argument payloadClass should be a string'
            );
        }

        if (!is_string($contentType)) {
            throw new \InvalidArgumentException(
                'Value for argument contentType should be a string'
            );
        }

        if (isset($this->payloadClassToContentTypeMap[$payloadClass])) {
            throw new \RuntimeException(
                'Content type for class ' . $payloadClass . ' was already set to ' . $this->payloadClassToContentTypeMap[$payloadClass]
            );
        }
        $this->payloadClassToContentTypeMap[$payloadClass] = $contentType;
    }
}