Consumer::consume() - Code Metrics - Inspection of "Merge pull request #20 from nepster-web/real_testi..." - nepster-web/php-simple-queue - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 5bd05b...9a2fe3 )

unknown

created 2021-03-29 13:52 UTC

Consumer::consume() A

↳ Parent: Consumer

Complexity

Conditions	6
Paths	8

Size

Total Lines	19
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	42

Importance

Changes	3
Bugs	1	Features	0

Metric	Value
cc	6
eloc	12
nc	8
nop	2
dl	0
loc	19
ccs	0
cts	11
cp	0
crap	42
rs	9.2222
c	3
b	1
f	0

<?php

declare(strict_types=1);

namespace Simple\Queue;

use Throwable;
use InvalidArgumentException;
use Simple\Queue\Store\StoreInterface;

/**
 * Class Consumer
 *
 * TODO: 1 - create job instance
 * TODO: 2 - move db operations to other class
 * TODO: 3 - dispatch job
 * TODO: 4 - deserialize data to job and processor
 *
 * @package Simple\Queue
 */
class Consumer
{
    /** Use to mark results as successful */
    public const STATUS_ACK = 'ACK';

    /** Use one more try if necessary */
    public const STATUS_REJECT = 'REJECT';

    /** Use in case of a delayed queue */
    public const STATUS_REQUEUE = 'REQUEUE';

    /** @var StoreInterface */
    protected StoreInterface $store;

    /** @var Producer */
    protected Producer $producer;

    /** @var Config|null */
    private ?Config $config;

    /**
     * Consumer constructor.
     * @param StoreInterface $store
     * @param Producer $producer
     * @param Config|null $config
     */
    public function __construct(StoreInterface $store, Producer $producer, ?Config $config = null)
    {
        $this->store = $store;
        $this->producer = $producer;
        $this->config = $config ?: Config::getDefault();
    }

    /**
     * The message has been successfully processed and will be removed from the queue
     *
     * @param Message $message
     */
    public function acknowledge(Message $message): void
    {
        $this->store->deleteMessage($message);
    }

    /**
     * Reject message with requeue option
     *
     * @param Message $message
     * @param bool $requeue
     */
    public function reject(Message $message, bool $requeue = false): void
    {
        $this->acknowledge($message);

        if ($requeue) {
            $redeliveryMessage = $this->producer->makeRedeliveryMessage($message);
            $this->producer->send($redeliveryMessage);
        }
    }

    /**
     * TODO: pass the processing message status to $eachCallback
     *
     * @param array $queues
     * @param callable|null $eachCallback
     */
    public function consume(array $queues = [], ?callable $eachCallback = null): void
    {
        $this->store->init();

        while (true) {
            if ($message = $this->store->fetchMessage($queues)) {
                try {
                    $this->processing($message);
                } catch (Throwable $throwable) {
                    try {
                        $this->processFailureResult($throwable, $message);
                    } catch (\Doctrine\DBAL\Exception $exception) {
                        // maybe lucky later
                    }
                }
                $eachCallback && $eachCallback($message, $throwable ?? null);
                continue;
            }
            usleep(200000); // 0.2 second
        }
    }

    /**
     * @param Message $message
     */
    protected function processing(Message $message): void
    {
        $this->store->changeMessageStatus($message, new Status(Status::IN_PROCESS));

        if ($message->isJob()) {
            try {
                $job = $this->config->getJob($message->getEvent());
                $job = $this->config->getJob(/** @scrutinizer ignore-type */ $message->getEvent());

                $result = $job->handle($message, $this->producer);

                $this->processSuccessResult($result, $message);
            } catch (Throwable $exception) {
                $this->processFailureResult($exception, $message);
            }

            return;
        }

        if ($this->config->hasProcessor($message->getQueue())) {
            try {
                $result = $this->config->getProcessor($message->getQueue())($message, $this->producer);

                $this->processSuccessResult($result, $message);
            } catch (Throwable $exception) {
                $this->processFailureResult($exception, $message);
            }

            return;
        }

        $this->processUndefinedHandlerResult($message);
    }

    /**
     * @param Message $message
     */
    protected function processUndefinedHandlerResult(Message $message): void
    {
        MessageHydrator::changeProperty($message, 'status', new Status(Status::UNDEFINED_HANDLER));
        MessageHydrator::changeProperty($message, 'error', 'Could not find any job or processor.');

        $this->reject($message, true);
    }

    /**
     * @param Throwable $exception
     * @param Message $message
     */
    protected function processFailureResult(Throwable $exception, Message $message): void
    {
        $newStatus = Status::REDELIVERED;

        $numberOfAttemptsBeforeFailure = $this->config->getNumberOfAttemptsBeforeFailure();

        if ($message->isJob()) {
            $job = $this->config->getJob($message->getEvent());
            $job = $this->config->getJob(/** @scrutinizer ignore-type */ $message->getEvent());
            if ($job->attempts()) {
class A
{
    function getObject()
    {
        return null;
    }

}

$a = new A();
if ($a->getObject()) {
                $numberOfAttemptsBeforeFailure = $job->attempts();
class A
{
    function getObject()
    {
        return null;
    }

}

$a = new A();
$object = $a->getObject();
            }
        }

        if (($message->getAttempts() + 1) >= $numberOfAttemptsBeforeFailure) {
            $newStatus = Status::FAILURE;
        }

        MessageHydrator::changeProperty($message, 'status', new Status($newStatus));
        MessageHydrator::changeProperty($message, 'error', (string)$exception);

        $this->reject($message, true);
    }

    /**
     * @param string $status
     * @param Message $message
     */
    protected function processSuccessResult(string $status, Message $message): void
    {
        if ($status === self::STATUS_ACK) {
            $this->acknowledge($message);

            return;
        }

        if ($status === self::STATUS_REJECT) {
            $this->reject($message);

            return;
        }

        if ($status === self::STATUS_REQUEUE) {
            $this->reject($message, true);

            return;
        }

        throw new InvalidArgumentException(sprintf('Unsupported result status: "%s".', $status));
    }
}


nepster-web / php-simple-queue

Push — master ( 5bd05b...9a2fe3 )

Consumer::consume() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like