Queue Class Reference

Inheritance diagram for Queue:

Public Member Functions
	__construct (Config $amqpConfig, EnvelopeFactory $envelopeFactory, $queueName, LoggerInterface $logger)
	acknowledge (EnvelopeInterface $envelope)
	subscribe ($callback)
	reject (EnvelopeInterface $envelope, $requeue=true, $rejectionMessage=null)
	push (EnvelopeInterface $envelope)
Public Member Functions inherited from QueueInterface
	dequeue ()

Detailed Description

Class Queue

@api

Since

100.0.0

Definition at line 22 of file Queue.php.

Constructor & Destructor Documentation

__construct()

__construct	(	Config	$amqpConfig,
		EnvelopeFactory	$envelopeFactory,
			$queueName,
		LoggerInterface	$logger
	)

Initialize dependencies.

Parameters

Config	$amqpConfig
EnvelopeFactory	$envelopeFactory
string	$queueName
LoggerInterface	$logger

Since

100.0.0

Definition at line 53 of file Queue.php.

      {
        $this->amqpConfig = $amqpConfig;
        $this->queueName = $queueName;
        $this->envelopeFactory = $envelopeFactory;
        $this->logger = $logger;
    }

Member Function Documentation

acknowledge()

acknowledge

(

EnvelopeInterface

$envelope

)

{Acknowledge message delivery

Parameters

EnvelopeInterface

$envelope

Returns

void

}

Since

100.0.0

Implements QueueInterface.

Definition at line 104 of file Queue.php.

    {
        $properties = $envelope->getProperties();
        $channel = $this->amqpConfig->getChannel();
        // @codingStandardsIgnoreStart
        try {
            $channel->basic_ack($properties['delivery_tag']);
        } catch (AMQPProtocolConnectionException $e) {
            throw new ConnectionLostException(
                $e->getMessage(),
                $e->getCode(),
                $e
            );
        }
        // @codingStandardsIgnoreEnd
    }

push()

push

(

EnvelopeInterface

$envelope

)

(Push message to queue directly, without using exchange

Parameters

EnvelopeInterface

$envelope

Returns

void

Since

100.1.0

)

Since

100.0.0

Implements QueueInterface.

Definition at line 178 of file Queue.php.

    {
        $messageProperties = $envelope->getProperties();
        $msg = new AMQPMessage(
            $envelope->getBody(),
            [
                'correlation_id' => $messageProperties['correlation_id'],
                'delivery_mode' => 2
            ]
        );
        $this->amqpConfig->getChannel()->basic_publish($msg, '', $this->queueName);
    }

reject()

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

(Reject message

Parameters

EnvelopeInterface	$envelope
bool	$requeue
string	$rejectionMessage

Returns

void

)

Since

100.0.0

Implements QueueInterface.

Definition at line 159 of file Queue.php.

    {
        $properties = $envelope->getProperties();

        $channel = $this->amqpConfig->getChannel();
        // @codingStandardsIgnoreStart
        $channel->basic_reject($properties['delivery_tag'], $requeue);
        // @codingStandardsIgnoreEnd
        if ($rejectionMessage !== null) {
            $this->logger->critical(
                new \Magento\Framework\Phrase('Message has been rejected: %message', ['message' => $rejectionMessage])
            );
        }
    }

subscribe()

subscribe

(

$callback

)

{Wait for messages and dispatch them

Parameters

callable | array

$callback

Returns

void

}

Since

100.0.0

Implements QueueInterface.

Definition at line 125 of file Queue.php.

    {
        $callbackConverter = function (AMQPMessage $message) use ($callback) {
            // @codingStandardsIgnoreStart
            $properties = array_merge(
                $message->get_properties(),
                [
                    'topic_name' => $message->delivery_info['routing_key'],
                    'delivery_tag' => $message->delivery_info['delivery_tag'],
                ]
            );
            // @codingStandardsIgnoreEnd
            $envelope = $this->envelopeFactory->create(['body' => $message->body, 'properties' => $properties]);

            if ($callback instanceof \Closure) {
                $callback($envelope);
            } else {
                call_user_func($callback, $envelope);
            }
        };

        $channel = $this->amqpConfig->getChannel();
        // @codingStandardsIgnoreStart
        $channel->basic_consume($this->queueName, '', false, false, false, false, $callbackConverter);
        // @codingStandardsIgnoreEnd
        while (count($channel->callbacks)) {
            $channel->wait();
        }
    }

---

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

• vendor/magento/framework-amqp/Queue.php