Queue Class Reference

Inheritance diagram for Queue:

Public Member Functions
	saveMessage ($messageTopic, $messageBody)
	saveMessages ($messageTopic, array $messages)
	linkQueues ($messageId, $queueNames)
	linkMessagesWithQueues (array $messageIds, array $queueNames)
	getMessages ($queueName, $limit=null)
	deleteMarkedMessages ()
	takeMessagesInProgress ($relationIds)
	pushBackForRetry ($relationId)
	changeStatus ($relationIds, $status)
Public Member Functions inherited from AbstractDb
	__construct (\Magento\Framework\Model\ResourceModel\Db\Context $context, $connectionName=null)
	__sleep ()
	__wakeup ()
	getIdFieldName ()
	getMainTable ()
	getTable ($tableName)
	getConnection ()
	load (\Magento\Framework\Model\AbstractModel $object, $value, $field=null)
	save (\Magento\Framework\Model\AbstractModel $object)
	delete (\Magento\Framework\Model\AbstractModel $object)
	addUniqueField ($field)
	resetUniqueField ()
	unserializeFields (\Magento\Framework\Model\AbstractModel $object)
	getUniqueFields ()
	hasDataChanged ($object)
	getChecksum ($table)
	afterLoad (\Magento\Framework\DataObject $object)
	beforeSave (\Magento\Framework\DataObject $object)
	afterSave (\Magento\Framework\DataObject $object)
	beforeDelete (\Magento\Framework\DataObject $object)
	afterDelete (\Magento\Framework\DataObject $object)
	serializeFields (\Magento\Framework\Model\AbstractModel $object)
Public Member Functions inherited from AbstractResource
	__construct ()
	getConnection ()
	beginTransaction ()
	addCommitCallback ($callback)
	commit ()
	rollBack ()
	getValidationRulesBeforeSave ()

Protected Member Functions
	_construct ()
	getQueueIdsByNames ($queueNames)
	getMessageStatusTable ()
	getQueueTable ()
	getMessageTable ()
Protected Member Functions inherited from AbstractDb
	_init ($mainTable, $idFieldName)
	_setResource ($connections, $tables=null)
	_setMainTable ($mainTable, $idFieldName=null)
	_getConnection ($resourceName)
	_getLoadSelect ($field, $value, $object)
	_initUniqueFields ()
	_prepareDataForSave (\Magento\Framework\Model\AbstractModel $object)
	_prepareValueForSave ($value, $type)
	_checkUnique (\Magento\Framework\Model\AbstractModel $object)
	_afterLoad (\Magento\Framework\Model\AbstractModel $object)
	_beforeSave (\Magento\Framework\Model\AbstractModel $object)
	_afterSave (\Magento\Framework\Model\AbstractModel $object)
	_beforeDelete (\Magento\Framework\Model\AbstractModel $object)
	_afterDelete (\Magento\Framework\Model\AbstractModel $object)
	_serializeFields (\Magento\Framework\Model\AbstractModel $object)
	prepareDataForUpdate ($object)
	isObjectNotNew (\Magento\Framework\Model\AbstractModel $object)
	saveNewObject (\Magento\Framework\Model\AbstractModel $object)
	updateObject (\Magento\Framework\Model\AbstractModel $object)
	processAfterSaves (\Magento\Framework\Model\AbstractModel $object)
	isModified (\Magento\Framework\Model\AbstractModel $object)
	processNotModifiedSave (\Magento\Framework\Model\AbstractModel $object)
Protected Member Functions inherited from AbstractResource
	_construct ()
	_serializeField (DataObject $object, $field, $defaultValue=null, $unsetEmpty=false)
	_unserializeField (DataObject $object, $field, $defaultValue=null)
	_prepareDataForTable (DataObject $object, $table)
	_prepareTableValueForSave ($value, $type)
	_getColumnsForEntityLoad (\Magento\Framework\Model\AbstractModel $object, $tableName)
	getSerializer ()

Additional Inherited Members
Protected Attributes inherited from AbstractDb
	$_resources
	$connectionName = \Magento\Framework\App\ResourceConnection::DEFAULT_CONNECTION
	$_connections = []
	$_resourceModel
	$_tables = []
	$_mainTable
	$_idFieldName
	$_isPkAutoIncrement = true
	$_useIsObjectNew = false
	$_mainTableFields
	$_uniqueFields = null
	$_serializableFields = []
	$transactionManager
	$objectRelationProcessor
Protected Attributes inherited from AbstractResource
	$serializer
	$_logger

Detailed Description

Resource model for queue.

Definition at line 13 of file Queue.php.

Member Function Documentation

_construct()

_construct ( )

protected

Model initialization

Returns

void

Definition at line 20 of file Queue.php.

    {
        $this->_init('queue', 'id');
    }

changeStatus()

changeStatus	(		$relationIds,
			$status
	)

Change message status.

Parameters

	int[]	$relationIds
int	$status

Returns

void

Definition at line 234 of file Queue.php.

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

deleteMarkedMessages()

deleteMarkedMessages

(

)

Delete messages if there is no queue whrere the message is not in status TO BE DELETED

Returns

void

Definition at line 168 of file Queue.php.

    {
        $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);
    }

getMessages()

getMessages	(		$queueName,
			$limit = null
	)

Retrieve messages from the specified queue.

Parameters

string	$queueName
int | null	$limit

Returns

array

Definition at line 128 of file Queue.php.

    {
        $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);
    }

getMessageStatusTable()

getMessageStatusTable ( )

protected

Get name of table storing message statuses and associations to queues.

Returns

string

Definition at line 248 of file Queue.php.

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

getMessageTable()

getMessageTable ( )

protected

Get name of table storing message body and topic.

Returns

string

Definition at line 268 of file Queue.php.

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

getQueueIdsByNames()

getQueueIdsByNames (   $queueNames )

protected

Retrieve array of queue IDs corresponding to the specified array of queue names.

Parameters

string[]

$queueNames

Returns

int[]

Definition at line 112 of file Queue.php.

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

getQueueTable()

getQueueTable ( )

protected

Get name of table storing declared queues.

Returns

string

Definition at line 258 of file Queue.php.

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

linkMessagesWithQueues()

linkMessagesWithQueues	(	array	$messageIds,
		array	$queueNames
	)

Add associations between the specified messages and queues.

Parameters

array	$messageIds
	string[]	$queueNames

Returns

$this

Definition at line 82 of file Queue.php.

    {
        $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;
    }

linkQueues()

linkQueues	(		$messageId,
			$queueNames
	)

Add associations between the specified message and queues.

Parameters

int	$messageId
	string[]	$queueNames

Returns

$this

Definition at line 70 of file Queue.php.

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

pushBackForRetry()

pushBackForRetry

(

$relationId

)

Set status of message to 'retry required' and increment number of processing trials.

Parameters

int

$relationId

Returns

void

Definition at line 215 of file Queue.php.

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

saveMessage()

saveMessage	(		$messageTopic,
			$messageBody
	)

Save message to 'queue_message' table.

Parameters

string	$messageTopic
string	$messageBody

Returns

int ID of the inserted record

Definition at line 32 of file Queue.php.

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

saveMessages()

saveMessages	(		$messageTopic,
		array	$messages
	)

Save messages in bulk to 'queue_message' table.

Parameters

string	$messageTopic
array	$messages

Returns

array List of IDs of inserted records

Definition at line 48 of file Queue.php.

    {
        $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);
    }

takeMessagesInProgress()

takeMessagesInProgress

(

$relationIds

)

Mark specified messages with 'in progress' status.

Parameters

int[]

$relationIds

Returns

int[] IDs of messages which should be taken in progress by current process.

If status was set to 'in progress' by some other process (due to race conditions), current process should not process the same message. So message will be processed only if current process was able to change its status.

Definition at line 188 of file Queue.php.

    {
        $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;
    }

---

The documentation for this class was generated from the following file:

• vendor/magento/module-mysql-mq/Model/ResourceModel/Queue.php