Queue.php

Go to the documentation of this file.

<?php
namespace Magento\MysqlMq\Model\ResourceModel;

use Magento\MysqlMq\Model\QueueManagement;

class Queue extends \Magento\Framework\Model\ResourceModel\Db\AbstractDb
{
    protected function _construct()
    {
        $this->_init('queue', 'id');
    }

    public function saveMessage($messageTopic, $messageBody)
    {
        $this->getConnection()->insert(
            $this->getMessageTable(),
            ['topic_name' => $messageTopic, 'body' => $messageBody]
        );
        return $this->getConnection()->lastInsertId($this->getMessageTable());
    }

    public function saveMessages($messageTopic, array $messages)
    {
        $data = [];
        foreach ($messages as $message) {
            $data[] = ['topic_name' => $messageTopic, 'body' => $message];
        }
        $rowCount = $this->getConnection()->insertMultiple($this->getMessageTable(), $data);
        $firstId = $this->getConnection()->lastInsertId($this->getMessageTable());
        $select = $this->getConnection()->select()
            ->from(['qm' => $this->getMessageTable()], ['id'])
            ->where('qm.id >= ?', $firstId)
            ->limit($rowCount);
        return $this->getConnection()->fetchCol($select);
    }

    public function linkQueues($messageId, $queueNames)
    {
        return $this->linkMessagesWithQueues([$messageId], $queueNames);
    }

    public function linkMessagesWithQueues(array $messageIds, array $queueNames)
    {
        $connection = $this->getConnection();
        $queueIds = $this->getQueueIdsByNames($queueNames);
        $data = [];
        foreach ($messageIds as $messageId) {
            foreach ($queueIds as $queueId) {
                $data[] = [
                    $queueId,
                    $messageId,
                    QueueManagement::MESSAGE_STATUS_NEW
                ];
            }
        }
        if (!empty($data)) {
            $connection->insertArray(
                $this->getMessageStatusTable(),
                ['queue_id', 'message_id', 'status'],
                $data
            );
        }
        return $this;
    }

    protected function getQueueIdsByNames($queueNames)
    {
        $selectObject = $this->getConnection()->select();
        $selectObject->from(['queue' => $this->getQueueTable()])
            ->columns(['id'])
            ->where('queue.name IN (?)', $queueNames);
        return $this->getConnection()->fetchCol($selectObject);
    }

    public function getMessages($queueName, $limit = null)
    {
        $connection = $this->getConnection();
        $select = $connection->select()
            ->from(
                ['queue_message' => $this->getMessageTable()],
                [QueueManagement::MESSAGE_TOPIC => 'topic_name', QueueManagement::MESSAGE_BODY => 'body']
            )->join(
                ['queue_message_status' => $this->getMessageStatusTable()],
                'queue_message.id = queue_message_status.message_id',
                [
                    QueueManagement::MESSAGE_QUEUE_RELATION_ID => 'id',
                    QueueManagement::MESSAGE_QUEUE_ID => 'queue_id',
                    QueueManagement::MESSAGE_ID => 'message_id',
                    QueueManagement::MESSAGE_STATUS => 'status',
                    QueueManagement::MESSAGE_UPDATED_AT => 'updated_at',
                    QueueManagement::MESSAGE_NUMBER_OF_TRIALS => 'number_of_trials'
                ]
            )->join(
                ['queue' => $this->getQueueTable()],
                'queue.id = queue_message_status.queue_id',
                [QueueManagement::MESSAGE_QUEUE_NAME => 'name']
            )->where(
                'queue_message_status.status IN (?)',
                [QueueManagement::MESSAGE_STATUS_NEW, QueueManagement::MESSAGE_STATUS_RETRY_REQUIRED]
            )->where('queue.name = ?', $queueName)
            ->order('queue_message_status.updated_at ASC');

        if ($limit) {
            $select->limit($limit);
        }

        return $connection->fetchAll($select);
    }

    public function deleteMarkedMessages()
    {
        $connection = $this->getConnection();

        $select = $connection->select()
            ->from(['queue_message_status' => $this->getMessageStatusTable()], ['message_id'])
            ->where('status <> ?', QueueManagement::MESSAGE_STATUS_TO_BE_DELETED)
            ->distinct();
        $messageIds = $connection->fetchCol($select);

        $condition = count($messageIds) > 0 ? ['id NOT IN (?)' => $messageIds] : null;
        $connection->delete($this->getMessageTable(), $condition);
    }

    public function takeMessagesInProgress($relationIds)
    {
        $takenMessagesRelationIds = [];
        foreach ($relationIds as $relationId) {
            $affectedRows = $this->getConnection()->update(
                $this->getMessageStatusTable(),
                ['status' => QueueManagement::MESSAGE_STATUS_IN_PROGRESS],
                ['id = ?' => $relationId]
            );
            if ($affectedRows) {
                $takenMessagesRelationIds[] = $relationId;
            }
        }
        return $takenMessagesRelationIds;
    }

    public function pushBackForRetry($relationId)
    {
        $this->getConnection()->update(
            $this->getMessageStatusTable(),
            [
                'status' => QueueManagement::MESSAGE_STATUS_RETRY_REQUIRED,
                'number_of_trials' => new \Zend_Db_Expr('number_of_trials+1')
            ],
            ['id = ?' => $relationId]
        );
    }

    public function changeStatus($relationIds, $status)
    {
        $this->getConnection()->update(
            $this->getMessageStatusTable(),
            ['status' => $status],
            ['id IN (?)' => $relationIds]
        );
    }

    protected function getMessageStatusTable()
    {
        return $this->getTable('queue_message_status');
    }

    protected function getQueueTable()
    {
        return $this->getTable('queue');
    }

    protected function getMessageTable()
    {
        return $this->getTable('queue_message');
    }
}