Config.php

Go to the documentation of this file.

<?php
namespace Magento\Framework\Amqp;

use Magento\Framework\Amqp\Connection\FactoryOptions;
use Magento\Framework\App\DeploymentConfig;
use Magento\Framework\App\ObjectManager;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Channel\AMQPChannel;
use Magento\Framework\Amqp\Connection\Factory as ConnectionFactory;

class Config
{
    const QUEUE_CONFIG = 'queue';

    const AMQP_CONFIG = 'amqp';

    const HOST = 'host';
    const PORT = 'port';
    const USERNAME = 'user';
    const PASSWORD = 'password';
    const VIRTUALHOST = 'virtualhost';
    const SSL = 'ssl';
    const SSL_OPTIONS = 'ssl_options';

    private $deploymentConfig;

    private $connection;

    private $channel;

    private $data;

    private $connectionName;

    private $connectionFactory;

    public function __construct(
        DeploymentConfig $config,
        $connectionName = 'amqp',
        ConnectionFactory $connectionFactory = null
    ) {
        $this->deploymentConfig = $config;
        $this->connectionName = $connectionName;
        $this->connectionFactory = $connectionFactory
            ?: ObjectManager::getInstance()->get(ConnectionFactory::class);
    }

    public function __destruct()
    {
        $this->closeConnection();
    }

    public function getValue($key)
    {
        $this->load();
        return isset($this->data[$key]) ? $this->data[$key] : null;
    }

    private function createConnection(): AbstractConnection
    {
        $sslEnabled = trim($this->getValue(self::SSL)) === 'true';
        $options = new FactoryOptions();
        $options->setHost($this->getValue(self::HOST));
        $options->setPort($this->getValue(self::PORT));
        $options->setUsername($this->getValue(self::USERNAME));
        $options->setPassword($this->getValue(self::PASSWORD));
        $options->setVirtualHost($this->getValue(self::VIRTUALHOST));
        $options->setSslEnabled($sslEnabled);
        if ($sslOptions = $this->getValue(self::SSL_OPTIONS)) {
            $options->setSslOptions($sslOptions);
        }

        return $this->connectionFactory->create($options);
    }

    public function getChannel()
    {
        if (!isset($this->connection) || !isset($this->channel)) {
            $this->connection = $this->createConnection();

            $this->channel = $this->connection->channel();
        }
        return $this->channel;
    }

    private function load()
    {
        if (null === $this->data) {
            $queueConfig = $this->deploymentConfig->getConfigData(self::QUEUE_CONFIG);
            if ($this->connectionName == self::AMQP_CONFIG) {
                $this->data = isset($queueConfig[self::AMQP_CONFIG]) ? $queueConfig[self::AMQP_CONFIG] : [];
            } else {
                $this->data = isset($queueConfig['connections'][$this->connectionName])
                    ? $queueConfig['connections'][$this->connectionName]
                    : [];
            }
            if (empty($this->data)) {
                throw  new \LogicException('Unknown connection name ' . $this->connectionName);
            }
        }
    }

    private function closeConnection()
    {
        if (isset($this->channel)) {
            $this->channel->close();
            unset($this->channel);
        }

        if (isset($this->connection)) {
            $this->connection->close();
            unset($this->connection);
        }
    }
}