diff --git a/vendor/magento/module-message-queue/Model/CheckIsAvailableMessagesInQueue.php b/vendor/magento/module-message-queue/Model/CheckIsAvailableMessagesInQueue.php index c097f461e621..49540e248319 100644 --- a/vendor/magento/module-message-queue/Model/CheckIsAvailableMessagesInQueue.php +++ b/vendor/magento/module-message-queue/Model/CheckIsAvailableMessagesInQueue.php @@ -7,6 +7,7 @@ namespace Magento\MessageQueue\Model; +use Magento\Framework\MessageQueue\CountableQueueInterface; use Magento\Framework\MessageQueue\QueueRepository; /** @@ -40,6 +41,9 @@ public function __construct(QueueRepository $queueRepository) public function execute($connectionName, $queueName) { $queue = $this->queueRepository->get($connectionName, $queueName); + if ($queue instanceof CountableQueueInterface) { + return $queue->count() > 0; + } $message = $queue->dequeue(); if ($message) { $queue->reject($message); diff --git a/vendor/magento/module-mysql-mq/Model/Driver/Queue.php b/vendor/magento/module-mysql-mq/Model/Driver/Queue.php index cbc2e951782f..6d29fc8aee57 100644 --- a/vendor/magento/module-mysql-mq/Model/Driver/Queue.php +++ b/vendor/magento/module-mysql-mq/Model/Driver/Queue.php @@ -5,16 +5,18 @@ */ namespace Magento\MysqlMq\Model\Driver; +use Magento\Framework\App\ObjectManager; +use Magento\Framework\MessageQueue\CountableQueueInterface; use Magento\Framework\MessageQueue\EnvelopeInterface; -use Magento\Framework\MessageQueue\QueueInterface; use Magento\MysqlMq\Model\QueueManagement; use Magento\Framework\MessageQueue\EnvelopeFactory; +use Magento\MysqlMq\Model\ResourceModel\Queue as QueueResourceModel; use Psr\Log\LoggerInterface; /** * Queue based on MessageQueue protocol */ -class Queue implements QueueInterface +class Queue implements CountableQueueInterface { /** * @var QueueManagement @@ -46,6 +48,11 @@ class Queue implements QueueInterface */ private $logger; + /** + * @var QueueResourceModel + */ + private $queueResourceModel; + /** * Queue constructor. * @@ -55,6 +62,7 @@ class Queue implements QueueInterface * @param string $queueName * @param int $interval * @param int $maxNumberOfTrials + * @param QueueResourceModel|null $queueResourceModel */ public function __construct( QueueManagement $queueManagement, @@ -62,7 +70,8 @@ public function __construct( LoggerInterface $logger, $queueName, $interval = 5, - $maxNumberOfTrials = 3 + $maxNumberOfTrials = 3, + ?QueueResourceModel $queueResourceModel = null ) { $this->queueManagement = $queueManagement; $this->envelopeFactory = $envelopeFactory; @@ -70,6 +79,8 @@ public function __construct( $this->interval = $interval; $this->maxNumberOfTrials = $maxNumberOfTrials; $this->logger = $logger; + $this->queueResourceModel = $queueResourceModel + ?? ObjectManager::getInstance()->get(QueueResourceModel::class); } /** @@ -151,4 +162,12 @@ public function push(EnvelopeInterface $envelope) [$this->queueName] ); } + + /** + * @inheritDoc + */ + public function count(): int + { + return $this->queueResourceModel->getMessagesCount($this->queueName); + } } diff --git a/vendor/magento/module-mysql-mq/Model/ResourceModel/Queue.php b/vendor/magento/module-mysql-mq/Model/ResourceModel/Queue.php index 2a45eafc63f2..a110f1efdd0c 100644 --- a/vendor/magento/module-mysql-mq/Model/ResourceModel/Queue.php +++ b/vendor/magento/module-mysql-mq/Model/ResourceModel/Queue.php @@ -5,6 +5,8 @@ */ namespace Magento\MysqlMq\Model\ResourceModel; +use Magento\Framework\DB\Select; +use Magento\Framework\DB\Sql\Expression; use Magento\MysqlMq\Model\QueueManagement; /** @@ -240,6 +242,35 @@ public function changeStatus($relationIds, $status) ); } + /** + * Get number of pending messages in the queue + * + * @param string $queueName + * @return int + */ + public function getMessagesCount(string $queueName): int + { + $connection = $this->getConnection(); + $select = $connection->select() + ->from( + ['queue_message' => $this->getMessageTable()], + )->join( + ['queue_message_status' => $this->getMessageStatusTable()], + 'queue_message.id = queue_message_status.message_id' + )->join( + ['queue' => $this->getQueueTable()], + 'queue.id = queue_message_status.queue_id' + )->where( + 'queue_message_status.status IN (?)', + [QueueManagement::MESSAGE_STATUS_NEW, QueueManagement::MESSAGE_STATUS_RETRY_REQUIRED] + )->where('queue.name = ?', $queueName); + + $select->reset(Select::COLUMNS); + $select->columns(new Expression('COUNT(*)')); + + return (int) $connection->fetchOne($select); + } + /** * Get name of table storing message statuses and associations to queues. * diff --git a/vendor/magento/framework-message-queue/CountableQueueInterface.php b/vendor/magento/framework-message-queue/CountableQueueInterface.php new file mode 100644 index 000000000000..a6aec418ab5d --- /dev/null +++ b/vendor/magento/framework-message-queue/CountableQueueInterface.php @@ -0,0 +1,26 @@ +