AbstractReceiver.php

См. документацию.

<?php
 
declare(strict_types=1);
 
namespace Bitrix\Main\Messenger\Receiver;
 
use Bitrix\Main\Application;
use Bitrix\Main\Diag\EventLogger;
use Bitrix\Main\Diag\ExceptionHandlerLog;
use Bitrix\Main\Messenger\Broker\BrokerInterface;
use Bitrix\Main\Messenger\Entity\MessageBox;
use Bitrix\Main\Messenger\Entity\MessageInterface;
use Bitrix\Main\Messenger\Internals\Exception\Broker\AckFailedException;
use Bitrix\Main\Messenger\Internals\Exception\Broker\BrokerReadException;
use Bitrix\Main\Messenger\Internals\Exception\Broker\RejectFailedException;
use Bitrix\Main\Messenger\Internals\Exception\Receiver\ProcessingException;
use Bitrix\Main\Messenger\Internals\Exception\Receiver\RecoverableMessageException;
use Bitrix\Main\Messenger\Internals\Exception\Receiver\UnprocessableMessageException;
use Bitrix\Main\Messenger\Internals\Exception\Receiver\UnrecoverableMessageException;
use Exception;

abstract class AbstractReceiver implements ReceiverInterface
{
    protected int $limit = 50;
 
    protected string $queueId;
 
    protected BrokerInterface $broker;
 
    public function setLimit(int $limit): self
    {
        $this->limit = $limit > 0 ? $limit : 50;
 
        return $this;
    }
 
    public function setQueueId(string $queueId): self
    {
        $this->queueId = $queueId;
 
        return $this;
    }
 
    public function setBroker(BrokerInterface $broker): self
    {
        $this->broker = $broker;
 
        return $this;
    }

    protected function getMessage(): ?MessageBox
    {
        return $this->broker->getOne($this->queueId);
    }

    protected function getMessages(): iterable
    {
        return $this->broker->get($this->queueId, $this->limit);
    }

    protected function ack(MessageBox $messageBox): void
    {
        $this->broker->ack($messageBox);
    }

    protected function reject(MessageBox $messageBox): void
    {
        $this->broker->reject($messageBox);
    }

    abstract protected function process(MessageInterface $message): void;

    public function run(): void
    {
        $logger = new EventLogger(
            'main',
            'MESSENGER_QUEUE',
            static function (array $context, string $message)
            {
                $messageBox = $context['message'] ?? null;
 
                if ($messageBox instanceof MessageBox)
                {
                    return [
                        'ITEM_ID' => $messageBox->getItemId(),
                        'DESCRIPTION' => sprintf(
                            '%s. Message: "%s" (%s). Queue: "%s". ItemId: "%s"',
                            $message,
                            $messageBox->getClassName(),
                            $messageBox->getId(),
                            $messageBox->getQueueId(),
                            $messageBox->getItemId()
                        )
                    ];
                }
 
                return [
                    'ITEM_ID' => null,
                ];
            }
        );
 
        $messageBoxes = $this->getMessages();

        foreach ($messageBoxes as $messageBox)
        {
            try
            {
                $this->process($messageBox->getMessage());
 
                $this->ack($messageBox);
            }
            catch (UnprocessableMessageException $e)
            {
                $this->reject($messageBox);
 
                Application::getInstance()->getExceptionHandler()->writeToLog(
                    $e,
                    ExceptionHandlerLog::CAUGHT_EXCEPTION
                );
            }
            catch (UnrecoverableMessageException $e)
            {
                $messageBox->kill();
 
                $this->reject($messageBox);
 
                $logger->notice(
                    sprintf(
                        'Message has unrecoverable case: "%s"',
                        $e->getMessage(),
                    ),
                    [
                        'message' => $messageBox,
                        'exception' => $e
                    ]
                );
            }
            catch (RecoverableMessageException $e)
            {
                $messageBox->requeue($e->getRetryDelay());
 
                $this->reject($messageBox);
 
                $logger->notice(
                    sprintf(
                        'Message has recoverable case: "%s"',
                        $e->getMessage(),
                    ),
                    [
                        'message' => $messageBox,
                        'exception' => $e
                    ]
                );
            }
            catch (Exception $e)
            {
                $e = new ProcessingException(
                    $messageBox,
                    $e->getMessage() . ' Message: ' . $messageBox->getId(),
                    $e->getCode(),
                    $e
                );
 
                $this->reject($messageBox);
 
                Application::getInstance()->getExceptionHandler()->writeToLog(
                    $e,
                    ExceptionHandlerLog::CAUGHT_EXCEPTION
                );
            }
        }
    }
}