MergedMessageProcessor.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\MessageQueue;

class MergedMessageProcessor implements MessageProcessorInterface
{
    private $messageStatusProcessor;

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

    public function process(
        QueueInterface $queue,
        ConsumerConfigurationInterface $configuration,
        array $messages,
        array $messagesToAcknowledge,
        array $mergedMessages
    ) {
        $this->messageStatusProcessor->acknowledgeMessages($queue, $messagesToAcknowledge);
        $this->dispatchMessages($queue, $configuration, $mergedMessages, $messages);
    }

    private function dispatchMessages(
        QueueInterface $queue,
        ConsumerConfigurationInterface $configuration,
        array $messageList,
        array $originalMessages
    ) {
        $originalMessagesIds = [];

        try {
            foreach ($messageList as $topicName => $messages) {
                foreach ($messages as $message) {
                    $callbacks = $configuration->getHandlers($topicName);
                    $originalMessagesIds = $message->getOriginalMessagesIds();

                    foreach ($callbacks as $callback) {
                        call_user_func($callback, $message->getMergedMessage());
                    }

                    $originalMessages = $this->getOriginalMessages($originalMessages, $originalMessagesIds);
                    $this->messageStatusProcessor->acknowledgeMessages($queue, $originalMessages);
                }
            }
        } catch (\Exception $e) {
            $originalMessages = $this->getOriginalMessages($originalMessages, $originalMessagesIds);
            $this->messageStatusProcessor->rejectMessages($queue, $originalMessages);
        }
    }

    private function getOriginalMessages(array $messages, array $messagesIds)
    {
        $originalMessages = [];

        foreach ($messagesIds as $messageId) {
            if (isset($messages[$messageId])) {
                $originalMessages[] = $messages[$messageId];
            }
        }

        return $originalMessages;
    }
}