How to delete a message in message_queue where the message is processed?

Question

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

<?php
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 {
            $this->execute($orderSearchResult);
            $this->_notifier->addMajor(
                __('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->_notifier->addCritical(
                $errorCode,
                $message
            );
            $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();
            $logger->addWriter($writer);
            $logger->info($orderItems);

        //$queue->deleteQueueByType($this->_type);
        //$queue->enqueueMultiRecords($orderCollectionArr);

        }
    }
}

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

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.