Bdf\QueueMessengerBundle\Transport\Handler\MessageBusHandler

Complexity

Total Complexity

10

Size/Duplication

Total Lines	95
Duplicated Lines	0 %

Test Coverage

Coverage

0%

Importance

Changes	1
Bugs	0	Features	0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	4	2
A	__invoke()	0	9	1
A	dispatch()	0	13	4
A	toEnvelope()	0	19	3

<?php

namespace Bdf\QueueMessengerBundle\Transport\Handler;

use Bdf\Queue\Message\EnvelopeInterface as QueuedEnvelope;
use Bdf\Queue\Message\InteractEnvelopeInterface;
use Bdf\QueueMessengerBundle\Transport\Stamp\NullStampsSerializer;
use Bdf\QueueMessengerBundle\Transport\Stamp\StampsSerializerInterface;
use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\LogicException;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Stamp\HandledStamp;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;

/**
 * Job for dispatch a queued command to the bus
 * If the message do not contains a valid object as data, an InternalMessage is dispatched
 *
 * If the a reply is requested on the message,
 * a synchronised dispatch is performed,
 * and the result is returned as reply
 */
class MessageBusHandler
{
    /**
     * The command dispatcher
     *
     * @var MessageBusInterface
     */
    private $dispatcher;

    /**
     * @var StampsSerializerInterface
     */
    private $stampsSerializer;


    /**
     * Constructor
     *
     * @param MessageBusInterface $dispatcher
     * @param StampsSerializerInterface|null $stampsSerializer
     */
    public function __construct(MessageBusInterface $dispatcher, StampsSerializerInterface $stampsSerializer = null)
    {
        $this->dispatcher = $dispatcher;
        $this->stampsSerializer = $stampsSerializer ?: new NullStampsSerializer();
    }

    /**
     * Handle a queued command
     *
     * @param mixed $message
     * @param QueuedEnvelope $queuedEnvelope
     */
    public function __invoke($message, QueuedEnvelope $queuedEnvelope)
    {
        $envelope = $this
            ->toEnvelope($message, $queuedEnvelope)
            // Symfony 4.3 add transport name in constructor
            ->with(new ReceivedStamp($queuedEnvelope->connection()->getName()))
        ;

        $this->dispatch($envelope, $queuedEnvelope);
    }

    /**
     * Dispatch the envelope to the bus
     *
     * If the message is replyable and a reply is requested, a synchronized call is performed and the result is returned
     *
     * @param Envelope $envelope
     * @param QueuedEnvelope $queuedEnvelope
     */
    private function dispatch(Envelope $envelope, QueuedEnvelope $queuedEnvelope): void
    {
        $envelope = $this->dispatcher->dispatch($envelope);

        if ($queuedEnvelope instanceof InteractEnvelopeInterface && $queuedEnvelope->message()->needsReply()) {
            /** @var HandledStamp[] $handledStamps */
            $handledStamps = $envelope->all(HandledStamp::class);

            if (!$handledStamps) {

The expression $handledStamps of type Symfony\Component\Messenger\Stamp\HandledStamp[] is implicitly converted to a boolean; are you sure this is intended? If so, consider using empty($expr) instead to make it clear that you intend to check for an array without elements.

                throw new LogicException(sprintf('Message of type "%s" was handled zero times. Exactly one handler is expected when using "%s::%s()".', \get_class($envelope->getMessage()), \get_class($this), __FUNCTION__));
            }

            $queuedEnvelope->reply($handledStamps[0]->getResult());
        }
    }

    /**
     * Get the envelope from the message payload
     *
     * @param mixed $message
     * @param QueuedEnvelope $queuedEnvelope
     *
     * @return Envelope
     */
    private function toEnvelope($message, QueuedEnvelope $queuedEnvelope): Envelope
    {
        if ($message instanceof Envelope) {
            return $message;
        }

        // TODO How to handle non object message
//        if (!is_object($message)) {
//            $message = new InternalMessage(
//                $queuedEnvelope->message()->name() ?: $queuedEnvelope->message()->queue(),
//                $message
//            );
//        }

        if ($stamps = $queuedEnvelope->message()->header('stamps')) {
            return new Envelope($message, $this->stampsSerializer->deserialize($stamps));
        }

        return new Envelope($message);
    }
}