PSB\Core\Pipeline\Outgoing\OutgoingPhysicalToDispatchConnector

Complexity

Total Complexity

7

Size/Duplication

Total Lines	76
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

100%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	9	1
A	invoke()	0	22	4
A	getStageContextClass()	0	4	1
A	getNextStageContextClass()	0	4	1

<?php
namespace PSB\Core\Pipeline\Outgoing;


use PSB\Core\DateTimeConverter;
use PSB\Core\HeaderTypeEnum;
use PSB\Core\Pipeline\Outgoing\StageContext\DispatchContext;
use PSB\Core\Pipeline\Outgoing\StageContext\OutgoingPhysicalMessageContext;
use PSB\Core\Pipeline\StageConnectorInterface;
use PSB\Core\Transport\TransportOperation;
use PSB\Core\Util\Clock\ClockInterface;

class OutgoingPhysicalToDispatchConnector implements StageConnectorInterface
{
    /**
     * @var OutgoingContextFactory
     */
    private $contextFactory;

    /**
     * @var DateTimeConverter
     */
    private $dateTimeConverter;

    /**
     * @var ClockInterface
     */
    private $clock;

    /**
     *
     * @param OutgoingContextFactory $contextFactory
     * @param DateTimeConverter      $dateTimeConverter
     * @param ClockInterface         $clock
     */
    public function __construct(
        OutgoingContextFactory $contextFactory,
        DateTimeConverter $dateTimeConverter,
        ClockInterface $clock
    ) {
        $this->contextFactory = $contextFactory;
        $this->dateTimeConverter = $dateTimeConverter;
        $this->clock = $clock;
    }

    /**
     * @param OutgoingPhysicalMessageContext $context
     * @param callable                       $next
     */
    public function invoke($context, callable $next)
    {
        $context->setHeader(HeaderTypeEnum::MESSAGE_ID, $context->getMessageId());
        $context->setHeader(
            HeaderTypeEnum::TIME_SENT,
            $this->dateTimeConverter->toWireFormattedString($this->clock->now())
        );

        $transportOperations = [];
        foreach ($context->getAddressTags() as $addressTag) {
            $transportOperations[] = new TransportOperation(clone $context->getMessage(), $addressTag);
        }

        $pendingOperations = $context->getPendingTransportOperations();

        if (!$context->isImmediateDispatchEnabled() && $pendingOperations) {
            $pendingOperations->addAll($transportOperations);
            return;
        }

        $next($this->contextFactory->createDispatchContext($transportOperations, $context));
    }

    /**
     * @return string
     */
    public static function getStageContextClass()
    {
        return OutgoingPhysicalMessageContext::class;
    }

    /**
     * @return string
     */
    public static function getNextStageContextClass()
    {
        return DispatchContext::class;
    }
}