yiisoft / queue-amqp

src/ExistingMessagesConsumer.php

<?php

declare(strict_types=1);

namespace Yiisoft\Queue\AMQP;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use Throwable;
use Yiisoft\Queue\Message\MessageInterface;

/**
 * @internal
 */
final class ExistingMessagesConsumer
{
    private bool $messageConsumed = false;

    public function __construct(
        private AMQPChannel $channel,
        private string $queueName,
        private MessageSerializerInterface $serializer
    ) {
    }

    /**
     * @param callable(MessageInterface): bool  $callback
     */
    public function consume(callable $callback): void
    {
        $this->channel->basic_consume(
            $this->queueName,
            '',
            false,
            false,
            false,
            false,
            function (AMQPMessage $amqpMessage) use ($callback): void {
                try {
                    $message = $this->serializer->unserialize($amqpMessage->body);

The property PhpAmqpLib\Message\AMQPMessage::$body has been deprecated: Will be removed in version 4.0, use getBody() instead.

                    if ($this->messageConsumed = $callback($message)) {
                        $this->channel->basic_ack($amqpMessage->getDeliveryTag());
                    }
                } catch (Throwable $exception) {
                    $this->messageConsumed = false;
                    $consumerTag = $amqpMessage->getConsumerTag();
                    if ($consumerTag !== null) {
                        $this->channel->basic_cancel($consumerTag);
                    }

                    throw $exception;
                }
            }
        );

        do {
            $this->messageConsumed = false;
            $this->channel->wait(null, true);
        } while ($this->messageConsumed === true);
    }
}