MessageProcessor.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\MessageQueue;

use Magento\Framework\App\ResourceConnection;

class MessageProcessor implements MessageProcessorInterface
{
    private $messageStatusProcessor;

    private $resource;

    public function __construct(
        MessageStatusProcessor $messageStatusProcessor,
        ResourceConnection $resource
    ) {
        $this->messageStatusProcessor = $messageStatusProcessor;
        $this->resource = $resource;
    }

    public function process(
        QueueInterface $queue,
        ConsumerConfigurationInterface $configuration,
        array $messages,
        array $messagesToAcknowledge,
        array $mergedMessages
    ) {
        try {
            $this->resource->getConnection()->beginTransaction();
            $this->messageStatusProcessor->acknowledgeMessages($queue, $messagesToAcknowledge);
            $this->dispatchMessages($configuration, $mergedMessages);
            $this->resource->getConnection()->commit();
            $this->messageStatusProcessor->acknowledgeMessages($queue, $messages);
        } catch (ConnectionLostException $e) {
            $this->resource->getConnection()->rollBack();
        } catch (\Exception $e) {
            $this->resource->getConnection()->rollBack();
            $this->messageStatusProcessor->rejectMessages($queue, $messages);
        }
    }

    private function dispatchMessages(ConsumerConfigurationInterface $configuration, array $messageList)
    {
        foreach ($messageList as $topicName => $messages) {
            foreach ($messages as $message) {
                $callbacks = $configuration->getHandlers($topicName);
                foreach ($callbacks as $callback) {
                    call_user_func($callback, $message);
                }
            }
        }
    }
}