Darkilliant\MqProcessBundle\Step\ConsumeMessageStep

Complexity

Total Complexity

16

Size/Duplication

Total Lines	102
Duplicated Lines	0 %

Test Coverage

Coverage

0%

Importance

Changes

0

12 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	3	1
A	getProvider()	0	3	1
A	valid()	0	3	1
A	count()	0	3	1
A	getCurrent()	0	12	3
A	execute()	0	5	1
A	configureOptionResolver()	0	9	1
A	getProgress()	0	3	1
A	getProviderName()	0	5	1
A	onFailedLoop()	0	7	2
A	next()	0	9	1
A	onSuccessLoop()	0	7	2

<?php

namespace Darkilliant\MqProcessBundle\Step;

use Darkilliant\MqProcessBundle\Message\Message;
use Darkilliant\MqProcessBundle\Message\Provider\AmqpMessageProvider;
use Darkilliant\MqProcessBundle\Message\Provider\MessageProviderFactory;
use Darkilliant\ProcessBundle\State\ProcessState;
use Darkilliant\ProcessBundle\Step\AbstractConfigurableStep;
use Darkilliant\ProcessBundle\Step\IterableStepInterface;
use Symfony\Component\OptionsResolver\OptionsResolver;

class ConsumeMessageStep extends AbstractConfigurableStep implements IterableStepInterface
{
    /** @var Message */
    private $currentMessage;

    /** @var MessageProviderFactory */
    private $providerFactory;

    /**
     * @throws \AMQPConnectionException
     */
    public function __construct(MessageProviderFactory $providerFactory)
    {
        $this->providerFactory = $providerFactory;
    }

    public function configureOptionResolver(OptionsResolver $resolver): OptionsResolver
    {
        $resolver->setRequired(['queue', 'exchange', 'client', 'ack_required', 'persistant', 'batch_count']);
        $resolver->setDefault('client', 'amqp_lib');
        $resolver->setDefault('ack_required', true);
        $resolver->setDefault('persistant', true);
        $resolver->setDefault('batch_count', 1);

        return parent::configureOptionResolver($resolver);
    }

    public function execute(ProcessState $state)
    {
        $state->info('start consume');
        $state->setContext('queue', $state->getOptions()['queue']);
        $this->getProvider($state)->consume($state->getOptions()['ack_required']);
    }

    public function getProgress(ProcessState $state)
    {
        return 50;
    }

    public function count(ProcessState $state)
    {
        return 100;
    }

    public function next(ProcessState $state)
    {
        $this->currentMessage = null;

        $raw = $this->getCurrent($state)->getBody();

        $data = json_decode($raw, true);
        $state->info('consume message', ['data' => $data]);
        $state->setData($data);
    }

    public function valid(ProcessState $state)
    {
        return true;
    }

    public function onSuccessLoop(ProcessState $state)
    {
        if (!$state->getOptions()['ack_required']) {
            return;
        }

        $this->getProvider($state)->messageOk($this->getCurrent($state));
    }

    public function onFailedLoop(ProcessState $state)
    {
        if (!$state->getOptions()['ack_required']) {
            return;
        }

        $this->getProvider($state)->messageKo($this->getCurrent($state));
    }

    private function getProviderName(ProcessState $state)
    {
        $options = $state->getOptions();

        return $options['queue'];
    }

    private function getProvider(ProcessState $state): AmqpMessageProvider
    {
        return $this->providerFactory->factory($this->getProviderName($state), $state->getOptions());
    }

    private function getCurrent(ProcessState $state): Message
    {
        while (null === $this->currentMessage) {
            $message = $this->getProvider($state)->fetchMessage();
            if (null !== $message) {
                return $this->currentMessage = $message;
            }

            $this->getProvider($state)->waitChannel(false);
        }

        return $this->currentMessage;
    }
}