TopicConfig.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\MessageQueue\Config\Reader\Xml\Converter;

use Magento\Framework\MessageQueue\ConfigInterface;
use Magento\Framework\MessageQueue\Config\Validator;
use Magento\Framework\Reflection\MethodsMap;
use Magento\Framework\Communication\ConfigInterface as CommunicationConfig;
use Magento\Framework\MessageQueue\ConfigInterface as QueueConfig;
use Magento\Framework\MessageQueue\ConsumerInterface;

class TopicConfig implements \Magento\Framework\Config\ConverterInterface
{
    const DEFAULT_TYPE = 'amqp';
    const DEFAULT_EXCHANGE = 'magento';
    const DEFAULT_INSTANCE = ConsumerInterface::class;

    private $xmlValidator;

    private $methodsMap;

    private $communicationConfig;

    public function __construct(
        MethodsMap $methodsMap,
        Validator $xmlValidator,
        CommunicationConfig $communicationConfig
    ) {
        $this->methodsMap = $methodsMap;
        $this->xmlValidator = $xmlValidator;
        $this->communicationConfig = $communicationConfig;
    }

    public function convert($source)
    {
        $topics = $this->extractTopics($source);
        $topics = $this->processWildcard($topics);
        $publishers = $this->buildPublishers($topics);
        $binds = $this->buildBinds($topics);
        $map = $this->buildExchangeTopicToQueue($topics);
        $consumers = $this->buildConsumers($topics);
        return [
            ConfigInterface::TOPICS => $this->buildTopicsConfiguration($topics),
            ConfigInterface::PUBLISHERS => $publishers,
            ConfigInterface::BINDS => $binds,
            ConfigInterface::CONSUMERS => $consumers,
            ConfigInterface::EXCHANGE_TOPIC_TO_QUEUES_MAP => $map,
        ];
    }

    private function buildTopicsConfiguration($topics)
    {
        $output = [];
        foreach ($topics as $topicName => $topicConfig) {
            $topicDefinition = $this->communicationConfig->getTopic($topicName);
            $schemaType =
                $topicDefinition['request_type'] == CommunicationConfig::TOPIC_REQUEST_TYPE_CLASS
                    ? QueueConfig::TOPIC_SCHEMA_TYPE_OBJECT
                    : QueueConfig::TOPIC_SCHEMA_TYPE_METHOD;
            $schemaValue = $topicDefinition[CommunicationConfig::TOPIC_REQUEST];
            $output[$topicName] = [
                'name' => $topicName,
                'schema' => [
                    'schema_type' => $schemaType,
                    'schema_value' => $schemaValue
                ],
                'response_schema' => [
                    'schema_type' => isset($topicDefinition['response']) ? QueueConfig::TOPIC_SCHEMA_TYPE_OBJECT : null,
                    'schema_value' => $topicDefinition['response']
                ],
                'is_synchronous' => $topicDefinition[CommunicationConfig::TOPIC_IS_SYNCHRONOUS],
                'publisher' => $topicConfig['type'] . '-' . $topicConfig['exchange']
            ];
        }
        return $output;
    }

    private function buildConsumers($topics)
    {
        $output = [];
        foreach ($topics as $topicName => $topicConfig) {
            $topic = $this->communicationConfig->getTopic($topicName);
            foreach ($topicConfig['queues'] as $queueName => $queueConfig) {
                $handlers = [];
                foreach ($queueConfig['handlers'] as $handler) {
                    if (!isset($handler[QueueConfig::CONSUMER_CLASS])) {
                        $handlerExploded = explode('::', $handler);
                        unset($handler);
                        $handler[QueueConfig::CONSUMER_CLASS] = $handlerExploded[0];
                        $handler[QueueConfig::CONSUMER_METHOD] = $handlerExploded[1];
                    }
                    $handlers[] = $handler;
                }
                $queueConfig['handlers'] = $handlers;

                $output[$queueConfig['consumer']] = [
                    'name' => $queueConfig['consumer'],
                    'queue' => $queueName,
                    'handlers' => [$topicName => $queueConfig['handlers']],
                    'instance_type' => $queueConfig['consumerInstance'] != null
                        ? $queueConfig['consumerInstance'] : self::DEFAULT_INSTANCE,
                    'consumer_type' => $topic[CommunicationConfig::TOPIC_IS_SYNCHRONOUS] ? 'sync' : 'async',
                    'max_messages' => $queueConfig['maxMessages'],
                    'connection' => $topicConfig['type']
                ];
            }
        }
        return $output;
    }

    private function processWildcard($topics)
    {
        $topicDefinitions = $this->communicationConfig->getTopics();
        $wildcardKeys = [];
        $topicNames = array_keys($topics);
        foreach ($topicNames as $topicName) {
            if (strpos($topicName, '*') !== false || strpos($topicName, '#') !== false) {
                $wildcardKeys[] = $topicName;
            }
        }
        foreach (array_unique($wildcardKeys) as $wildcardKey) {
            $pattern = $this->xmlValidator->buildWildcardPattern($wildcardKey);
            foreach (array_keys($topicDefinitions) as $topicName) {
                if (preg_match($pattern, $topicName)) {
                    if (isset($topics[$topicName])) {
                        $topics[$topicName] = array_merge($topics[$topicName], $topics[$wildcardKey]);
                    } else {
                        $topics[$topicName] = $topics[$wildcardKey];
                    }
                }
            }
            unset($topics[$wildcardKey]);
        }
        return $topics;
    }

    private function buildPublishers($topics)
    {
        $output = [];
        foreach ($topics as $topicConfig) {
            $publisherName = $topicConfig['type'] . '-' . $topicConfig['exchange'];
            $output[$publisherName] = [
                'name' => $publisherName,
                'connection' => $topicConfig['type'],
                'exchange' => $topicConfig['exchange']
            ];
        }
        return $output;
    }

    private function buildBinds($topics)
    {
        $output = [];
        foreach ($topics as $topicName => $topicConfig) {
            $queueNames = array_keys($topicConfig['queues']);
            foreach ($queueNames as $queueName) {
                $name = $topicName . '--' . $topicConfig['exchange']. '--' .$queueName;
                $output[$name] = [
                    'queue' => $queueName,
                    'exchange' => $topicConfig['exchange'],
                    'topic' => $topicName,
                ];
            }
        }
        return $output;
    }

    private function buildExchangeTopicToQueue($topics)
    {
        $output = [];
        foreach ($topics as $topicName => $topicConfig) {
            $key = $topicConfig['type'] . '-' . $topicConfig['exchange'] . '--' . $topicName;
            $queueNames = array_keys($topicConfig['queues']);
            foreach ($queueNames as $queueName) {
                $output[$key][] = $queueName;
                $output[$key] = array_unique($output[$key]);
            }
        }
        return $output;
    }

    private function extractTopics($config)
    {
        $output = [];
        foreach ($config->getElementsByTagName('broker') as $brokerNode) {
            $topicName = $this->getAttributeValue($brokerNode, 'topic');
            $output[$topicName] = [
                ConfigInterface::TOPIC_NAME => $topicName,
                'type' => $this->getAttributeValue($brokerNode, 'type', self::DEFAULT_TYPE),
                'exchange' => $this->getAttributeValue($brokerNode, 'exchange', self::DEFAULT_EXCHANGE),
                'consumerInstance' => $this->getAttributeValue($brokerNode, 'consumerInstance'),
                'maxMessages' => $this->getAttributeValue($brokerNode, 'maxMessages'),
                'queues' => $this->extractQueuesFromBroker($brokerNode, $topicName)
            ];
        }
        return $output;
    }

    protected function extractQueuesFromBroker(\DOMElement $brokerNode, $topicName)
    {
        $queues = [];
        $topicConfig = $this->communicationConfig->getTopic($topicName);
        foreach ($brokerNode->getElementsByTagName('queue') as $queueNode) {
            $handler = $this->getAttributeValue($queueNode, 'handler');
            $queueName = $this->getAttributeValue($queueNode, 'name');
            $queue = [
               'name'=> $queueName,
               'handlerName' => $this->getAttributeValue($queueNode, 'handlerName'),
               'handlers' => $handler ? ['default' => $handler] : $topicConfig['handlers'],
               'exchange' => $this->getAttributeValue($queueNode, 'exchange'),
               'consumer' => $this->getAttributeValue($queueNode, 'consumer'),
               'consumerInstance' => $this->getAttributeValue($queueNode, 'consumerInstance'),
               'maxMessages' => $this->getAttributeValue($queueNode, 'maxMessages', null),
               'type' => $this->getAttributeValue($queueNode, 'type')

            ];
            $queues[$queueName] = $queue;
        }
        return $queues;
    }

    protected function getAttributeValue(\DOMNode $node, $attributeName, $default = null)
    {
        $item = $node->attributes->getNamedItem($attributeName);
        return $item ? $item->nodeValue : $default;
    }
}