ConsumerFactory.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\MessageQueue;

use Magento\Framework\MessageQueue\ConfigInterface as QueueConfig;
use Magento\Framework\Exception\LocalizedException;
use Magento\Framework\ObjectManagerInterface;
use Magento\Framework\Phrase;
use Magento\Framework\MessageQueue\ConsumerInterface;
use Magento\Framework\MessageQueue\Consumer\ConfigInterface as ConsumerConfig;
use Magento\Framework\MessageQueue\Consumer\Config\ConsumerConfigItemInterface;
use Magento\Framework\Communication\ConfigInterface as CommunicationConfig;

class ConsumerFactory
{
    private $objectManager = null;

    private $consumerConfig;

    private $communicationConfig;

    public function __construct(
        QueueConfig $queueConfig,
        ObjectManagerInterface $objectManager
    ) {
        $this->objectManager = $objectManager;
    }

    public function get($consumerName, $batchSize = 0)
    {
        $consumerConfig = $this->getConsumerConfig()->getConsumer($consumerName);
        if ($consumerConfig === null) {
            throw new LocalizedException(
                new Phrase('Specified consumer "%consumer" is not declared.', ['consumer' => $consumerName])
            );
        }

        return $this->objectManager->create(
            $consumerConfig->getConsumerInstance(),
            [
                'configuration' => $this->createConsumerConfiguration($consumerConfig),
                'batchSize' => $batchSize,
            ]
        );
    }

    private function createConsumerConfiguration($consumerConfigItem)
    {
        $customConsumerHandlers = [];
        foreach ($consumerConfigItem->getHandlers() as $handlerConfig) {
            $customConsumerHandlers[] = [
                $this->objectManager->create($handlerConfig->getType()),
                $handlerConfig->getMethod()
            ];
        }
        $topics = [];
        foreach ($this->getCommunicationConfig()->getTopics() as $topicConfig) {
            $topicName = $topicConfig[CommunicationConfig::TOPIC_NAME];
            $topics[$topicName] = [
                ConsumerConfigurationInterface::TOPIC_HANDLERS => $customConsumerHandlers
                    ?: $this->getHandlersFromCommunicationConfig($topicName),
                ConsumerConfigurationInterface::TOPIC_TYPE => $topicConfig[CommunicationConfig::TOPIC_IS_SYNCHRONOUS]
                    ? ConsumerConfiguration::TYPE_SYNC
                    : ConsumerConfiguration::TYPE_ASYNC
            ];
        }
        $configData = [
            ConsumerConfigurationInterface::CONSUMER_NAME => $consumerConfigItem->getName(),
            ConsumerConfigurationInterface::QUEUE_NAME => $consumerConfigItem->getQueue(),
            ConsumerConfigurationInterface::TOPICS => $topics,
            ConsumerConfigurationInterface::MAX_MESSAGES => $consumerConfigItem->getMaxMessages(),
        ];

        return $this->objectManager->create(
            \Magento\Framework\MessageQueue\ConsumerConfiguration::class,
            ['data' => $configData]
        );
    }

    private function getConsumerConfig()
    {
        if ($this->consumerConfig === null) {
            $this->consumerConfig = \Magento\Framework\App\ObjectManager::getInstance()->get(ConsumerConfig::class);
        }
        return $this->consumerConfig;
    }

    private function getCommunicationConfig()
    {
        if ($this->communicationConfig === null) {
            $this->communicationConfig = \Magento\Framework\App\ObjectManager::getInstance()
                ->get(CommunicationConfig::class);
        }
        return $this->communicationConfig;
    }

    private function getHandlersFromCommunicationConfig($topicName)
    {
        $topicConfig = $this->getCommunicationConfig()->getTopic($topicName);
        $handlers = [];
        foreach ($topicConfig[CommunicationConfig::TOPIC_HANDLERS] as $handlerConfig) {
            $handlers[] = [
                $this->objectManager->create($handlerConfig[CommunicationConfig::HANDLER_TYPE]),
                $handlerConfig[CommunicationConfig::HANDLER_METHOD]
            ];
        }
        return $handlers;
    }
}