Innmind\AMQP\Client\Channel\Basic\Get\GetOk

Complexity

Total Complexity

9

Size/Duplication

Total Lines	90
Duplicated Lines	0 %

Test Coverage

Coverage

100%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	20	1
A	requeue()	0	6	1
B	__invoke()	0	33	6
A	reject()	0	6	1

<?php
declare(strict_types = 1);

namespace Innmind\AMQP\Client\Channel\Basic\Get;

use Innmind\AMQP\{
    Client\Channel\Basic\Get,
    Model\Basic\Get as Command,
    Model\Basic\Message,
    Model\Basic\Ack,
    Model\Basic\Reject as RejectCommand,
    Transport\Connection,
    Transport\Frame\Channel,
    Exception\Reject,
    Exception\Requeue,
};

final class GetOk implements Get
{
    private $connection;
    private $channel;
    private $command;
    private $message;
    private $deliveryTag;
    private $redelivered;
    private $exchange;
    private $routingKey;
    private $messageCount;
    private $consumed = false;

    public function __construct(
        Connection $connection,
        Channel $channel,
        Command $command,
        Message $message,
        int $deliveryTag,
        bool $redelivered,
        string $exchange,
        string $routingKey,
        int $messageCount
    ) {
        $this->connection = $connection;
        $this->channel = $channel;
        $this->command = $command;
        $this->message = $message;
        $this->deliveryTag = $deliveryTag;
        $this->redelivered = $redelivered;
        $this->exchange = $exchange;
        $this->routingKey = $routingKey;
        $this->messageCount = $messageCount;
    }

    /**
     * {@inheritdoc}
     */
    public function __invoke(callable $consume): void
    {
        if ($this->consumed) {
            return;
        }

        try {
            $consume(
                $this->message,
                $this->redelivered,
                $this->exchange,
                $this->routingKey,
                $this->messageCount
            );

            if (!$this->command->shouldAutoAcknowledge()) {
                $this
                    ->connection
                    ->send($this->connection->protocol()->basic()->ack(
                        $this->channel,
                        new Ack($this->deliveryTag)
                    ));
            }

            $this->consumed = true;
        } catch (Reject $e) {
            $this->reject();
        } catch (Requeue $e) {
            $this->requeue();
        } catch (\Throwable $e) {
            $this->reject();

            throw $e;
        }
    }

    private function reject(): void
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->reject(
                $this->channel,
                new RejectCommand($this->deliveryTag)
            )
        );
    }

    private function requeue(): void
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->reject(
                $this->channel,
                RejectCommand::requeue($this->deliveryTag)
            )
        );
    }
}