PSB\Core\Transport\RabbitMq\RabbitMqMessagePusher

Complexity

Total Complexity

4

Size/Duplication

Total Lines	71
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Test Coverage

Coverage

68.18%

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	1
A	init()	0	9	2
A	start()	0	16	1

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


use PSB\Core\EndpointControlToken;
use PSB\Core\Transport\MessagePusherInterface;
use PSB\Core\Transport\PushPipe;
use PSB\Core\Transport\PushSettings;
use PSB\Core\Transport\ReceiveCancellationToken;

class RabbitMqMessagePusher implements MessagePusherInterface
{
    /**
     * @var BrokerModel
     */
    private $brokerModel;

    /**
     * @var MessageProcessor
     */
    private $messageProcessor;

    /**
     * @var PushPipe
     */
    private $pushPipe;

    /**
     * @var PushSettings
     */
    private $pushSettings;

    /**
     * @param BrokerModel      $brokerModel
     * @param MessageProcessor $messageProcessor
     */
    public function __construct(
        BrokerModel $brokerModel,
        MessageProcessor $messageProcessor
    ) {
        $this->brokerModel = $brokerModel;
        $this->messageProcessor = $messageProcessor;
    }

    /**
     * It initializes the message pusher with the pipe it will send messages through.
     * The pipe is invoked for each incoming message.
     *
     * @param PushPipe     $pushPipe
     * @param PushSettings $pushSettings
     */
    public function init(PushPipe $pushPipe, PushSettings $pushSettings)
    {
        $this->pushPipe = $pushPipe;
        $this->pushSettings = $pushSettings;

        if ($pushSettings->isPurgeOnStartup()) {
            $this->brokerModel->purgeQueue($pushSettings->getInputQueue());
        }
    }

    /**
     * Starts pushing messages
     */
    public function start()
    {
        $this->brokerModel->consume(
            $this->pushSettings->getInputQueue(),
            function (\AMQPEnvelope $envelope, \AMQPQueue $queue) {
                return $this->messageProcessor->process(
                    $envelope,
                    $queue,
                    $this->pushPipe,
                    $this->pushSettings->getErrorQueue(),
                    new ReceiveCancellationToken(),
                    new EndpointControlToken()
                );
            }
        );
    }
}