PSB\Core\Transport\RabbitMq\RabbitMqMessageDispatcher

Complexity

Total Complexity

7

Size/Duplication

Total Lines	83
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	8

Test Coverage

Coverage

100%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	9	1
A	dispatch()	0	17	4
A	sendMessage()	0	10	1
A	publishMessage()	0	10	1

<?php
namespace PSB\Core\Transport\RabbitMq;


use PSB\Core\Exception\InvalidArgumentException;
use PSB\Core\Routing\MulticastAddressTag;
use PSB\Core\Routing\UnicastAddressTag;
use PSB\Core\Transport\MessageDispatcherInterface;
use PSB\Core\Transport\OutgoingPhysicalMessage;
use PSB\Core\Transport\TransportOperations;

class RabbitMqMessageDispatcher implements MessageDispatcherInterface
{
    /**
     * @var RoutingTopology
     */
    private $routingTopology;

    /**
     * @var BrokerModel
     */
    private $brokerModel;

    /**
     * @var MessageConverter
     */
    private $messageConverter;

    /**
     * @param RoutingTopology $routingTopology
     * @param BrokerModel $brokerModel
     * @param MessageConverter $messageConverter
     */
    public function __construct(
        RoutingTopology $routingTopology,
        BrokerModel $brokerModel,
        MessageConverter $messageConverter
    ) {
        $this->routingTopology = $routingTopology;
        $this->brokerModel = $brokerModel;
        $this->messageConverter = $messageConverter;
    }

    /**
     * @param TransportOperations $transportOperations
     */
    public function dispatch(TransportOperations $transportOperations)
    {
        foreach ($transportOperations->getTransportOperations() as $transportOperation) {
            $addressTag = $transportOperation->getAddressTag();
            $message = $transportOperation->getMessage();
            if ($addressTag instanceof UnicastAddressTag) {
                $this->sendMessage($message, $addressTag);
            } elseif ($addressTag instanceof MulticastAddressTag) {
                $this->publishMessage($message, $addressTag);
            } else {
                $tagType = get_class($addressTag);
                throw new InvalidArgumentException(
                    "Transport operations contain an unsupported type of '$tagType'. Supported types are 'PSB\Core\\Routing\\UnicastAddressTag' and 'PSB\Core\\Routing\\MulticastAddressTag'."
                );
            }
        }
    }

    /**
     * @param OutgoingPhysicalMessage $message
     * @param UnicastAddressTag       $addressTag
     */
    private function sendMessage(OutgoingPhysicalMessage $message, UnicastAddressTag $addressTag)
    {
        $attributes = $this->messageConverter->composeRabbitMqAttributes($message);
        $this->routingTopology->send(
            $this->brokerModel,
            $addressTag->getDestination(),
            $message->getBody(),
            $attributes
        );
    }

    /**
     * @param OutgoingPhysicalMessage $message
     * @param MulticastAddressTag     $addressTag
     */
    private function publishMessage(OutgoingPhysicalMessage $message, MulticastAddressTag $addressTag)
    {
        $attributes = $this->messageConverter->composeRabbitMqAttributes($message);
        $this->routingTopology->publish(
            $this->brokerModel,
            $addressTag->getMessageType(),
            $message->getBody(),
            $attributes
        );
    }
}