I'm try to create a message queue example by find some example on internet (i.e https://store.magenest.com/blog/create-a-message-queue-in-magento-2/), here is the consumer class

namespace Magenest\Salesforce\Model\Queue;

use Magento\Framework\Exception\LocalizedException;

use Magento\Framework\Filesystem;
use Magento\Framework\Notification\NotifierInterface;

 * Class Consumer
 * @package Magenest\Salesforce\Model\Queue
class Consumer
    /* @var \Psr\Log\LoggerInterface  */
    protected $_logger;

    /* @var \Magento\Sales\Api\OrderRepositoryInterface  */
    protected $_orderRepository;

    /* @var NotifierInterface  */
    protected $_notifier;

     * @var Filesystem
    private $filesystem;

    /* @var \Magento\Framework\Serialize\Serializer\Json  */
    protected $_json;

    /* @var string  */
    protected $_type = null;

    /* @var \Magento\Framework\Message\ManagerInterface  */
    protected $_messageManager;

     * Consumer constructor.
     * @param \Psr\Log\LoggerInterface $logger
     * @param \Magento\Sales\Api\OrderRepositoryInterface $orderRepository
     * @param NotifierInterface $notifier
     * @param Filesystem $filesystem
     * @param \Magento\Framework\Serialize\Serializer\Json $json
    public function __construct(
        \Psr\Log\LoggerInterface $logger,
        \Magento\Sales\Api\OrderRepositoryInterface $orderRepository,
        NotifierInterface $notifier,
        \Magento\Framework\Message\ManagerInterface $messageManager,
        Filesystem $filesystem,
        \Magento\Framework\Serialize\Serializer\Json $json
        $this->_logger = $logger;
        $this->_orderRepository = $orderRepository;
        $this->_notifier = $notifier;
        $this->_messageManager = $messageManager;
        $this->filesystem = $filesystem;
        $this->_json = $json;

     * @param string $orderSearchResult
    public function process($orderSearchResult)
        try {
                __('Your queue are ready'),
                __('You can check your orders at Salesforce Queue page')
        } catch (\Exception $e) {
            $errorCode = $e->getCode();
            $message = __('Sorry, something went wrong during add order to queue. Please see log for details.');
            $this->_logger->critical($errorCode .": ". $message);

     * @param $orderItems
     * @throws LocalizedException
    private function execute($orderItems)
        $orderCollectionArr = [];
        /* @var \Magenest\Salesforce\Model\Queue $queue */
        //$queue = $this->_queueFactory->create();

        $orderItems = $this->_json->unserialize($orderItems);

        if(is_array($orderItems)) {

            foreach ($orderItems as $type => $orderId) {
                $this->_type = $type;
                $orderCollectionArr[] = [
                    'type' => $type,
                    'entity_id' => $orderId,
                    'priority' => 1,

            $writer = new \Zend_Log_Writer_Stream(BP . '/var/log/testQueue.log');
            $logger = new \Zend_Log();



the class queueFactory is missing in the tutorial, the main things is how to delete the message after the message is processed?

1 Answer 1


Generally you do nothing. Messages in queue shouldn't be deleted by you after it has been processed. That should have been abstracted away and handled by Magento. Note that if you're using RabbitMQ, messages are removed from the queue once when the consumer acks it.

The idea here is that you publish messages to the queue and consumers pick it up and delegate it to a handler which will process the message. If the handler fails for any reason, the consumer will mark the message for retries or dead letter it if it cannot be processed anymore.

In the article you've shared above, messages are indeed published to a queue and then published to somewhere else.

In your handler, you do not have to do anything. As long as it executes your execute function entirely, that message is considered to be successfully processed.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.