DbBroker.php

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

<?php
 
declare(strict_types=1);
 
namespace Bitrix\Main\Messenger\Internals\Broker;
 
use Bitrix\Main\ArgumentException;
use Bitrix\Main\DI\ServiceLocator;
use Bitrix\Main\Messenger\Broker\BrokerInterface;
use Bitrix\Main\Messenger\Entity\MessageInterface;
use Bitrix\Main\Messenger\Entity\MessageBox;
use Bitrix\Main\Messenger\Entity\ProcessingParam\ProcessingParamInterface;
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\Broker\SendFailedException;
use Bitrix\Main\Messenger\Internals\Exception\Storage\PersistenceException;
use Bitrix\Main\Messenger\Internals\Config\QueueConfigRegistry;
use Bitrix\Main\Messenger\Internals\Retry\MultiplierRetryStrategy;
use Bitrix\Main\Messenger\Internals\Retry\RetryStrategyInterface;
use Bitrix\Main\Messenger\Internals\Storage\Db\DbStorage;
use Bitrix\Main\Messenger\Internals\Storage\StorageInterface;
use Bitrix\Main\ORM\Entity;
use Bitrix\Main\SystemException;
 
final class DbBroker implements BrokerInterface
{
    public const TYPE_CODE = 'db';
 
    private readonly StorageInterface $storage;
 
    private MultiplierRetryStrategy $retryStrategy;
 
    public function __construct(
        Entity $tableEntity,
        RetryStrategyInterface $retryStrategy = new MultiplierRetryStrategy()
    )
    {
        $this->storage = new DbStorage($tableEntity);
        $this->retryStrategy = $retryStrategy;
 
        $this->storage->unlockStaleMessages();
    }

    public function send(MessageInterface $message, string $queueId, array $params = []): void
    {
        $retryStrategy = $this->getRetryStrategy($queueId);
 
        $messageBox = new MessageBox($message);
 
        $messageBox
            ->setQueueId($queueId)
            ->setTtl($retryStrategy->getMaxRetryCount())
        ;
 
        $this->addProcessingParams($messageBox, $params);
 
        try
        {
            $this->retry([$this->storage, 'save'], $messageBox);
        }
        catch (PersistenceException $e)
        {
            throw new SendFailedException($messageBox, $e);
        }
    }
 
    public function get(string $queueId, int $limit = 50): iterable
    {
        try
        {
            return $this->storage->getReadyMessagesOfQueue($queueId, $limit);
        }
        catch (PersistenceException $e)
        {
            throw new BrokerReadException(
                'Unable to lock messages: ' . $e->getMessage(),
                $e->getCode(),
                $e
            );
        }
        catch (SystemException $e)
        {
            throw new BrokerReadException(
                'Unable to read messages: ' . $e->getMessage(),
                $e->getCode(),
                $e
            );
        }
    }
 
    public function getOne(string $queueId): ?MessageBox
    {
        try
        {
            return $this->storage->getOneByQueue($queueId);
        }
        catch (PersistenceException $e)
        {
            throw new BrokerReadException(
                'Unable to lock message: ' . $e->getMessage(),
                $e->getCode(),
                $e
            );
        }
        catch (SystemException $e)
        {
            throw new BrokerReadException(
                'Unable to read message: ' . $e->getMessage(),
                $e->getCode(),
                $e
            );
        }
    }

    public function ack(MessageBox $message): void
    {
        try
        {
            $this->retry([$this->storage, 'delete'], $message);
        }
        catch (PersistenceException $e)
        {
            throw new AckFailedException($message, $e);
        }
    }

    public function reject(MessageBox $message): void
    {
        $message->reject();
 
        try
        {
            if ($message->isDied())
            {
                $this->retry([$this->storage, 'delete'], $message);
 
                return;
            }
 
            $this->retry([$this->storage, 'save'], $message);
        }
        catch (PersistenceException $e)
        {
            throw new RejectFailedException($message, $e);
        }
    }

    private function retry(callable $callable, MessageBox $messageBox): void
    {
        $retry = 1;
 
        while (true)
        {
            try
            {
                call_user_func($callable, $messageBox);
 
                return;
            }
            catch (PersistenceException $e)
            {
                if ($retry > $this->retryStrategy->getMaxRetryCount() - 1)
                {
                    throw $e;
                }
 
                usleep($this->retryStrategy->getWaitingTime($retry) * 1000);
 
                $retry++;
            }
        }
    }

    private function addProcessingParams(MessageBox $messageBox, array $params): void
    {
        foreach ($params as $param)
        {
            if (!$param instanceof ProcessingParamInterface)
            {
                throw new ArgumentException(
                    sprintf('The type of message processing params should be "%s"', ProcessingParamInterface::class)
                );
            }
 
            $messageBox = $param->applyTo($messageBox);
        }
    }
 
    private function getRetryStrategy(string $queueId): RetryStrategyInterface
    {
        return ServiceLocator::getInstance()
            ->get(QueueConfigRegistry::class)
            ->getQueueConfig($queueId)
            ->retryStrategy;
    }
}