PSB\Core\ErrorHandling\FirstLevelRetry\Pipeline\FirstLevelRetryPipelineStep

Complexity

Total Complexity

7

Size/Duplication

Total Lines	65
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

100%

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	invoke()	0	28	5
A	getStageContextClass()	0	4	1

<?php
namespace PSB\Core\ErrorHandling\FirstLevelRetry\Pipeline;


use PSB\Core\ErrorHandling\FirstLevelRetry\FirstLevelRetryHeaderTypeEnum;
use PSB\Core\ErrorHandling\FirstLevelRetry\FirstLevelRetryPolicy;
use PSB\Core\ErrorHandling\FirstLevelRetry\FirstLevelRetryStorage;
use PSB\Core\Exception\CriticalErrorException;
use PSB\Core\Exception\MessageDeserializationException;
use PSB\Core\Pipeline\Incoming\StageContext\TransportReceiveContext;
use PSB\Core\Pipeline\PipelineStepInterface;

class FirstLevelRetryPipelineStep implements PipelineStepInterface
{
    /**
     * @var FirstLevelRetryStorage
     */
    private $retryStorage;

    /**
     * @var FirstLevelRetryPolicy
     */
    private $retryPolicy;

    /**
     * @param FirstLevelRetryStorage $retryStorage
     * @param FirstLevelRetryPolicy  $retryPolicy
     */
    public function __construct(FirstLevelRetryStorage $retryStorage, FirstLevelRetryPolicy $retryPolicy)
    {
        $this->retryStorage = $retryStorage;
        $this->retryPolicy = $retryPolicy;
    }

    /**
     * @param TransportReceiveContext $context
     * @param callable                $next
     *
     * @throws \Throwable
     */
    public function invoke($context, callable $next)
    {
        try {
            $next();
        } catch (CriticalErrorException $e) {
            // no retry for critical errors
            throw $e;
        } catch (MessageDeserializationException $e) {
            // no retry for invalid messages
            throw $e;
        } catch (\Throwable $t) {
            $messageId = $context->getMessageId();

            $numberOfRetries = $this->retryStorage->getFailuresForMessage($messageId);

            if ($this->retryPolicy->shouldGiveUp($numberOfRetries)) {
                $this->retryStorage->clearFailuresForMessage($messageId);

                $message = $context->getMessage();
                $message->setHeader(FirstLevelRetryHeaderTypeEnum::RETRIES, $numberOfRetries);

                throw $t;
            }

            $this->retryStorage->incrementFailuresForMessage($context->getMessageId());
            $context->abortReceiveOperation();
        }
    }

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