TopologyInstaller.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\Amqp;

use Magento\Framework\Amqp\Topology\ExchangeInstaller;
use Magento\Framework\Amqp\Topology\QueueInstaller;
use Magento\Framework\MessageQueue\Topology\ConfigInterface;

class TopologyInstaller
{
    private $topologyConfig;

    private $exchangeInstaller;

    private $configPool;

    private $queueInstaller;

    private $connectionTypeResolver;

    protected $logger;

    public function __construct(
        ConfigInterface $topologyConfig,
        ExchangeInstaller $exchangeInstaller,
        ConfigPool $configPool,
        QueueInstaller $queueInstaller,
        ConnectionTypeResolver $connectionTypeResolver,
        \Psr\Log\LoggerInterface $logger
    ) {
        $this->topologyConfig = $topologyConfig;
        $this->exchangeInstaller = $exchangeInstaller;
        $this->configPool = $configPool;
        $this->queueInstaller = $queueInstaller;
        $this->connectionTypeResolver = $connectionTypeResolver;
        $this->logger = $logger;
    }

    public function install()
    {
        try {
            foreach ($this->topologyConfig->getQueues() as $queue) {
                if ($this->connectionTypeResolver->getConnectionType($queue->getConnection()) != 'amqp') {
                    continue;
                }
                $amqpConfig = $this->configPool->get($queue->getConnection());
                $this->queueInstaller->install($amqpConfig->getChannel(), $queue);
            }
            foreach ($this->topologyConfig->getExchanges() as $exchange) {
                if ($this->connectionTypeResolver->getConnectionType($exchange->getConnection()) != 'amqp') {
                    continue;
                }
                $amqpConfig = $this->configPool->get($exchange->getConnection());
                $this->exchangeInstaller->install($amqpConfig->getChannel(), $exchange);
            }
        } catch (\Exception $e) {
            $this->logger->error("AMQP topology installation failed: {$e->getMessage()}\n{$e->getTraceAsString()}");
        }
    }
}