Queue Class Reference

Inheritance diagram for Queue:

Public Member Functions
	__construct (QueueManagement $queueManagement, EnvelopeFactory $envelopeFactory, LoggerInterface $logger, $queueName, $interval=5, $maxNumberOfTrials=3)
	dequeue ()
	acknowledge (EnvelopeInterface $envelope)
	subscribe ($callback)
	reject (EnvelopeInterface $envelope, $requeue=true, $rejectionMessage=null)
	push (EnvelopeInterface $envelope)

Detailed Description

Queue based on MessageQueue protocol

Definition at line 17 of file Queue.php.

Constructor & Destructor Documentation

__construct()

__construct	(	QueueManagement	$queueManagement,
		EnvelopeFactory	$envelopeFactory,
		LoggerInterface	$logger,
			$queueName,
			$interval = 5,
			$maxNumberOfTrials = 3
	)

Queue constructor.

Parameters

QueueManagement	$queueManagement
EnvelopeFactory	$envelopeFactory
LoggerInterface	$logger
string	$queueName
int	$interval
int	$maxNumberOfTrials

Definition at line 59 of file Queue.php.

      {
        $this->queueManagement = $queueManagement;
        $this->envelopeFactory = $envelopeFactory;
        $this->queueName = $queueName;
        $this->interval = $interval;
        $this->maxNumberOfTrials = $maxNumberOfTrials;
        $this->logger = $logger;
    }

Member Function Documentation

acknowledge()

acknowledge

(

EnvelopeInterface

$envelope

)

{Acknowledge message delivery

Parameters

EnvelopeInterface

$envelope

Returns

void

}

Implements QueueInterface.

Definition at line 97 of file Queue.php.

    {
        $properties = $envelope->getProperties();
        $relationId = $properties[QueueManagement::MESSAGE_QUEUE_RELATION_ID];

        $this->queueManagement->changeStatus($relationId, QueueManagement::MESSAGE_STATUS_COMPLETE);
    }

dequeue()

dequeue

(

)

{Get message from queue

Returns

EnvelopeInterface

}

{Get message from queue

Returns

EnvelopeInterface

}

Since

100.0.0

Implements QueueInterface.

Definition at line 78 of file Queue.php.

    {
        $envelope = null;
        $messages = $this->queueManagement->readMessages($this->queueName, 1);
        if (isset($messages[0])) {
            $properties = $messages[0];

            $body = $properties[QueueManagement::MESSAGE_BODY];
            unset($properties[QueueManagement::MESSAGE_BODY]);

            $envelope = $this->envelopeFactory->create(['body' => $body, 'properties' => $properties]);
        }

        return $envelope;
    }

push()

push

(

EnvelopeInterface

$envelope

)

Push message to queue directly, without using exchange

Parameters

EnvelopeInterface

$envelope

Returns

void

Since

100.1.0

Implements QueueInterface.

Definition at line 144 of file Queue.php.

    {
        $properties = $envelope->getProperties();
        $this->queueManagement->addMessageToQueues(
            $properties[QueueManagement::MESSAGE_TOPIC],
            $envelope->getBody(),
            [$this->queueName]
        );
    }

reject()

reject	(	EnvelopeInterface	$envelope,
			$requeue = true,
			$rejectionMessage = null
	)

{Reject message

Parameters

EnvelopeInterface	$envelope
bool	$requeue
string	$rejectionMessage

Returns

void

}

Implements QueueInterface.

Definition at line 126 of file Queue.php.

    {
        $properties = $envelope->getProperties();
        $relationId = $properties[QueueManagement::MESSAGE_QUEUE_RELATION_ID];

        if ($properties[QueueManagement::MESSAGE_NUMBER_OF_TRIALS] < $this->maxNumberOfTrials && $requeue) {
            $this->queueManagement->pushToQueueForRetry($relationId);
        } else {
            $this->queueManagement->changeStatus([$relationId], QueueManagement::MESSAGE_STATUS_ERROR);
            if ($rejectionMessage !== null) {
                $this->logger->critical(__('Message has been rejected: %1', $rejectionMessage));
            }
        }
    }

subscribe()

subscribe

(

$callback

)

{Wait for messages and dispatch them

Parameters

callable | array

$callback

Returns

void

}

Implements QueueInterface.

Definition at line 108 of file Queue.php.

    {
        while (true) {
            while ($envelope = $this->dequeue()) {
                try {
                    call_user_func($callback, $envelope);
                    $this->acknowledge($envelope);
                } catch (\Exception $e) {
                    $this->reject($envelope);
                }
            }
            sleep($this->interval);
        }
    }

---

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

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