Werkspot\MessageQueue\DeliveryQueue\Amqp\AmqpMessageHandler::handle()

Complexity

Conditions	5
Paths	35

Size

Total Lines	45
Code Lines	24

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	23
CRAP Score	5

Importance

Changes

0

<?php
    public function handle(AMQPMessage $amqpMessage): void
    {
        $this->initiateHandling($amqpMessage);
        $this->registerShutdownHandler();

        // RabbitMQ can and WILL send duplicate messages to clients. You are recommended to handle message in an
        // idempotent way, but for us this is not really possible, because sending out emails is never idempotent.
        // So we are doing the next best thing, deduplicating them.
        //
        // When deduplicating them we should do it as early as possible and in the most performing way as possible to
        // prevent race conditions where 2 consumers get the message at near the same time. So make sure we deduplicate
        // as soon as possible to keep the duplicate executing at a minimum.
        //
        // See: https://www.rabbitmq.com/reliability.html#consumer
        //  > [..] messages can be duplicated, and consumers must be prepared to handle them. If possible, ensure that
        //  > your consumers handle messages in an idempotent way rather than explicitly deal with deduplication.

        try {
            $this->claim($amqpMessage);
        } catch (AlreadyClaimedException $e) {
            $this->logAlreadyClaimedMessage($amqpMessage);
            // If we already processed it, we should still ack it here, otherwise rabbit will try to do it again, which
            // is exactly what we don't want.
            $this->acknowledgeMessage($amqpMessage);
            $this->terminateHandling();
            return;
        } catch (CanNotClaimException $e) {
            $this->logger->notice($e->getMessage());
            // if we don't have a message ID we can not claim it nor deduplicate it so
            // we should always process it, just to be sure we don't lose a message
        }

        try {
            $this->handler->handle($this->unpack($amqpMessage));
            $this->acknowledgeMessage($amqpMessage);
        } catch (AmqpMessageUnpackingException $e) {
            $this->logger->warning($e->getMessage());
            $this->persistenceClient->persistUndeliverableMessage($amqpMessage->body, $e->getMessage());
            $this->acknowledgeMessage($amqpMessage);
        } catch (UnrecoverablePersistenceException $e) {
            $this->logger->warning($e->getMessage());
            $this->acknowledgeMessage($amqpMessage);
            throw $e;
        } finally {
            $this->terminateHandling();
        }