QueueManagement.php

Go to the documentation of this file.

<?php
namespace Magento\MysqlMq\Model;

class QueueManagement
{
    const MESSAGE_TOPIC = 'topic_name';
    const MESSAGE_BODY = 'body';
    const MESSAGE_ID = 'message_id';
    const MESSAGE_STATUS = 'status';
    const MESSAGE_UPDATED_AT = 'updated_at';
    const MESSAGE_QUEUE_ID = 'queue_id';
    const MESSAGE_QUEUE_NAME = 'queue_name';
    const MESSAGE_QUEUE_RELATION_ID = 'relation_id';
    const MESSAGE_NUMBER_OF_TRIALS = 'retries';

    const MESSAGE_STATUS_NEW = 2;
    const MESSAGE_STATUS_IN_PROGRESS = 3;
    const MESSAGE_STATUS_COMPLETE= 4;
    const MESSAGE_STATUS_RETRY_REQUIRED = 5;
    const MESSAGE_STATUS_ERROR = 6;
    const MESSAGE_STATUS_TO_BE_DELETED = 7;

    const XML_PATH_SUCCESSFUL_MESSAGES_LIFETIME = 'system/mysqlmq/successful_messages_lifetime';
    const XML_PATH_FAILED_MESSAGES_LIFETIME = 'system/mysqlmq/failed_messages_lifetime';
    const XML_PATH_RETRY_IN_PROGRESS_AFTER = 'system/mysqlmq/retry_inprogress_after';
    const XML_PATH_NEW_MESSAGES_LIFETIME = 'system/mysqlmq/new_messages_lifetime';
    private $messageResource;

    private $scopeConfig;

    private $dateTime;

    private $messageStatusCollectionFactory;

    public function __construct(
        \Magento\MysqlMq\Model\ResourceModel\Queue $messageResource,
        \Magento\Framework\App\Config\ScopeConfigInterface $scopeConfig,
        \Magento\MysqlMq\Model\ResourceModel\MessageStatusCollectionFactory $messageStatusCollectionFactory,
        \Magento\Framework\Stdlib\DateTime\DateTime $dateTime
    ) {
        $this->messageResource = $messageResource;
        $this->scopeConfig = $scopeConfig;
        $this->dateTime = $dateTime;
        $this->messageStatusCollectionFactory = $messageStatusCollectionFactory;
    }

    public function addMessageToQueues($topic, $message, $queueNames)
    {
        $messageId = $this->messageResource->saveMessage($topic, $message);
        $this->messageResource->linkQueues($messageId, $queueNames);
        return $this;
    }

    public function addMessagesToQueues($topic, $messages, $queueNames)
    {
        $messageIds = $this->messageResource->saveMessages($topic, $messages);
        $this->messageResource->linkMessagesWithQueues($messageIds, $queueNames);
        return $this;
    }

    public function markMessagesForDelete()
    {
        $collection = $this->messageStatusCollectionFactory->create()
            ->addFieldToFilter(
                'status',
                ['in' => $this->getStatusesToClear()]
            );

        foreach ($collection as $messageStatus) {
            $this->processMessagePerStatus($messageStatus);
        }

        $this->messageResource->deleteMarkedMessages();
    }

    private function processMessagePerStatus($messageStatus)
    {
        $now = $this->dateTime->gmtTimestamp();

        if ($messageStatus->getStatus() == self::MESSAGE_STATUS_COMPLETE
            && strtotime($messageStatus->getUpdatedAt()) < ($now - $this->getCompletedMessageLifetime())) {
            $messageStatus->setStatus(self::MESSAGE_STATUS_TO_BE_DELETED)
                ->save();
        } elseif ($messageStatus->getStatus() == self::MESSAGE_STATUS_ERROR
            && strtotime($messageStatus->getUpdatedAt()) < ($now - $this->getErrorMessageLifetime())) {
            $messageStatus->setStatus(self::MESSAGE_STATUS_TO_BE_DELETED)
                ->save();
        } elseif ($messageStatus->getStatus() == self::MESSAGE_STATUS_IN_PROGRESS
            && strtotime($messageStatus->getUpdatedAt()) < ($now - $this->getInProgressRetryAfter())
        ) {
            $this->pushToQueueForRetry($messageStatus->getId());
        } elseif ($messageStatus->getStatus() == self::MESSAGE_STATUS_NEW
            && strtotime($messageStatus->getUpdatedAt()) < ($now - $this->getNewMessageLifetime())
        ) {
            $messageStatus->setStatus(self::MESSAGE_STATUS_TO_BE_DELETED)
                ->save();
        }
    }

    private function getStatusesToClear()
    {
        $statusesToDelete = [];
        if ($this->getCompletedMessageLifetime() > 0) {
            $statusesToDelete[] = self::MESSAGE_STATUS_COMPLETE;
        }

        if ($this->getErrorMessageLifetime() > 0) {
            $statusesToDelete[] = self::MESSAGE_STATUS_ERROR;
        }

        if ($this->getNewMessageLifetime() > 0) {
            $statusesToDelete[] = self::MESSAGE_STATUS_NEW;
        }

        if ($this->getInProgressRetryAfter() > 0) {
            $statusesToDelete[] = self::MESSAGE_STATUS_IN_PROGRESS;
        }
        return $statusesToDelete;
    }

    private function getCompletedMessageLifetime()
    {
        return 60 * (int)$this->scopeConfig->getValue(
            self::XML_PATH_SUCCESSFUL_MESSAGES_LIFETIME,
            \Magento\Store\Model\ScopeInterface::SCOPE_STORE
        );
    }

    private function getErrorMessageLifetime()
    {
        return  60 * (int)$this->scopeConfig->getValue(
            self::XML_PATH_FAILED_MESSAGES_LIFETIME,
            \Magento\Store\Model\ScopeInterface::SCOPE_STORE
        );
    }

    private function getInProgressRetryAfter()
    {
        return 60 * (int)$this->scopeConfig->getValue(
            self::XML_PATH_RETRY_IN_PROGRESS_AFTER,
            \Magento\Store\Model\ScopeInterface::SCOPE_STORE
        );
    }

    private function getNewMessageLifetime()
    {
        return  60 * (int)$this->scopeConfig->getValue(
            self::XML_PATH_NEW_MESSAGES_LIFETIME,
            \Magento\Store\Model\ScopeInterface::SCOPE_STORE
        );
    }

    public function readMessages($queue, $maxMessagesNumber = null)
    {
        $selectedMessages = $this->messageResource->getMessages($queue, $maxMessagesNumber);
        /* The logic below allows to prevent the same message being processed by several consumers in parallel */
        $selectedMessagesRelatedIds = [];
        foreach ($selectedMessages as &$message) {
            /* Set message status here to avoid extra reading from DB after it is updated */
            $message[self::MESSAGE_STATUS] = self::MESSAGE_STATUS_IN_PROGRESS;
            $selectedMessagesRelatedIds[] = $message[self::MESSAGE_QUEUE_RELATION_ID];
        }
        $takenMessagesRelationIds = $this->messageResource->takeMessagesInProgress($selectedMessagesRelatedIds);
        if (count($selectedMessages) == count($takenMessagesRelationIds)) {
            return $selectedMessages;
        } else {
            $selectedMessages = array_combine($selectedMessagesRelatedIds, array_values($selectedMessages));
            return array_intersect_key($selectedMessages, array_flip($takenMessagesRelationIds));
        }
    }

    public function pushToQueueForRetry($messageRelationId)
    {
        $this->messageResource->pushBackForRetry($messageRelationId);
    }

    public function changeStatus($messageRelationIds, $status)
    {
        $this->messageResource->changeStatus($messageRelationIds, $status);
    }
}