Shippinno\Job\Infrastructure\Ui\Console\Laravel\Command\JobConsume

Complexity

Total Complexity

12

Size/Duplication

Total Lines	122
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	9

Test Coverage

Coverage

64.29%

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	12	1
A	createConsumer()	0	7	1
B	consume()	0	31	6
A	delayMessage()	0	8	2
A	handle()	0	7	2

<?php

namespace Shippinno\Job\Infrastructure\Ui\Console\Laravel\Command;

use Doctrine\ORM\EntityManager;
use Illuminate\Console\Command;
use Illuminate\Contracts\Console\Application;
use Illuminate\Contracts\Container\Container;
use Interop\Queue\PsrConsumer;
use Interop\Queue\PsrContext;
use Interop\Queue\PsrMessage;
use JMS\Serializer\Serializer;
use Shippinno\Job\Application\Job\Job;
use Shippinno\Job\Application\Job\JobRunner;
use Shippinno\Job\Domain\Model\JobFailedException;

class JobConsume extends Command
{
    /**
     * {@inheritdoc}
     */
    protected $signature = 'job:consume';

    /**
     * @var PsrContext
     */
    private $context;

    /**
     * @var Serializer
     */
    private $serializer;

    /**
     * @var EntityManager
     */
    private $entityManager;

    /**
     * @var Container
     */
    private $container;

    /**
     * @param PsrContext $context
     * @param Serializer $serializer
     * @param EntityManager $entityManager
     * @param Container $container
     */
    public function __construct(
        PsrContext $context,
        Serializer $serializer,
        EntityManager $entityManager,

You have injected the EntityManager via parameter $entityManager. This is generally not recommended as it might get closed and become unusable. Instead, it is recommended to inject the ManagerRegistry and retrieve the EntityManager via getManager() each time you need it.

        Container $container
    ) {
        parent::__construct();
        $this->context = $context;
        $this->serializer = $serializer;
        $this->entityManager = $entityManager;
        $this->container = $container;
    }

    /**
     * @throws \Doctrine\Common\Persistence\Mapping\MappingException
     * @throws \Doctrine\ORM\ORMException
     * @throws \Doctrine\ORM\OptimisticLockException
     */
    public function handle()
    {
        $consumer = $this->createConsumer();
        while (true) {
            $this->consume($consumer);
        }
    }

    /**
     * @return PsrConsumer
     */
    protected function createConsumer(): PsrConsumer
    {
        $queue = $this->context->createQueue('test');
        $consumer = $this->context->createConsumer($queue);

        return $consumer;
    }

    /**
     * @param PsrConsumer $consumer
     * @throws \Doctrine\Common\Persistence\Mapping\MappingException
     * @throws \Doctrine\ORM\ORMException
     * @throws \Doctrine\ORM\OptimisticLockException
     */
    protected function consume(PsrConsumer $consumer)
    {
        $this->entityManager->clear();
        $message = $consumer->receive();
        if (null === $message) {
            return;
        }
        $messageBody = json_decode($message->getBody());
        /** @var Job $job */
        $job = $this->serializer->deserialize($messageBody->body, $messageBody->name, 'json');
        if ($message->getProperty('attempts') > $job->maxAttempts()) {
            $consumer->reject($message);
        }
        /** @var JobRunner $jobRunner */
        $jobRunner = $this->container->make($job->jobRunner());
        try {
            $jobRunner->run($job);
            $consumer->acknowledge($message);
        } catch (JobFailedException $e) {
            $message->setProperty(
                'attempts',
                intval($message->getProperty('attempts', 0)) ?: 0
            );
            if ($job->reattemptDelay() > 0) {
                $message = $this->delayMessage($message, $job->reattemptDelay());
            }
            $consumer->reject($message, true);
        } finally {
            $this->entityManager->flush();
        }
    }

    /**
     * @param PsrMessage $message
     * @param int $delay
     * @return PsrMessage
     */
    protected function delayMessage(PsrMessage $message, int $delay)
    {
        if (method_exists($message, 'setDelaySeconds')) {
            $message->setDelaySeconds($delay);

The method setDelaySeconds() does not seem to exist on object<Interop\Queue\PsrMessage>.

        }

        return $message;
    }
}