ConsumerConfiguration.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\MessageQueue;

use Magento\Framework\MessageQueue\ConfigInterface as MessageQueueConfig;
use Magento\Framework\MessageQueue\Consumer\ConfigInterface as ConsumerConfig;
use Magento\Framework\Communication\ConfigInterface as CommunicationConfig;

class ConsumerConfiguration implements ConsumerConfigurationInterface
{
    const CONSUMER_TYPE = "consumer_type";

    const HANDLERS = 'handlers';

    private $data;

    private $queueRepository;

    private $consumerConfig;

    private $communicationConfig;

    public function __construct(QueueRepository $queueRepository, MessageQueueConfig $messageQueueConfig, $data = [])
    {
        $this->data = $data;
        $this->queueRepository = $queueRepository;
    }

    public function getConsumerName()
    {
        return $this->getData(self::CONSUMER_NAME);
    }

    public function getMaxMessages()
    {
        return $this->getData(self::MAX_MESSAGES);
    }

    public function getQueueName()
    {
        return $this->getData(self::QUEUE_NAME);
    }

    public function getType()
    {
        $topics = $this->getData(self::TOPICS);
        if (count($topics) > 1) {
            throw new \LogicException(
                'Current method is deprecated and does not support more than 1 topic declarations for consumer. '
                . 'Use \Magento\Framework\MessageQueue\ConsumerConfiguration::getConsumerType instead. '
                . "Multiple topics declared for consumer '{$this->getConsumerName()}'"
            );
        } elseif (count($topics) < 1) {
            throw new \LogicException(
                "There must be at least one topic declared for consumer '{$this->getConsumerName()}'."
            );
        }
        // Get the only topic and read consumer type from its declaration. Necessary for backward compatibility
        $topicConfig = reset($topics);
        return $topicConfig[self::TOPIC_TYPE];
    }

    public function getHandlers($topicName)
    {
        return $this->getTopicConfig($topicName)[self::TOPIC_HANDLERS];
    }

    public function getTopicNames()
    {
        $topics = $this->getData(self::TOPICS);
        return is_array($topics) && count($topics) ? array_keys($topics) : [];
    }

    public function getQueue()
    {
        $connectionName = $this->getConsumerConfig()->getConsumer($this->getConsumerName())->getConnection();
        return $this->queueRepository->get($connectionName, $this->getQueueName());
    }

    public function getMessageSchemaType($topicName)
    {
        return $this->getCommunicationConfig()->getTopic($topicName)[CommunicationConfig::TOPIC_REQUEST_TYPE];
    }

    private function getTopicConfig($topicName)
    {
        if (!isset($this->getData(self::TOPICS)[$topicName])) {
            throw new \LogicException("Consumer configuration for {$topicName} topic not found.");
        }
        return $this->getData(self::TOPICS)[$topicName];
    }

    private function getData($key)
    {
        if (!isset($this->data[$key])) {
            return null;
        }
        return $this->data[$key];
    }

    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;
    }
}