Alchemy\Queue\Amqp\AmqpMessageQueueFactory

Complexity

Total Complexity

13

Size/Duplication

Total Lines	112
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	2

Test Coverage

Coverage

95%

Importance

Changes	2
Bugs	0	Features	1

7 Methods

Rating	Name	Duplication	Size	Complexity
A	create()	0	4	1
A	__construct()	0	4	2
A	getNamedQueue()	0	6	1
A	buildQueue()	0	15	2
A	declareExchange()	0	16	3
A	getConnection()	0	9	2
A	getChannel()	0	8	2

<?php

namespace Alchemy\Queue\Amqp;

use Alchemy\Queue\MessageQueue;
use Alchemy\Queue\MessageQueueFactory;

class AmqpMessageQueueFactory implements MessageQueueFactory
{
    /**
     * @param array $configuration
     * @return AmqpMessageQueueFactory
     */
    public static function create(array $configuration = [])
    {
        return new self(AmqpConfiguration::parse($configuration));
    }

    /**
     * @var AmqpConfiguration
     */
    private $configuration;

    /**
     * @var \AMQPExchange
     */
    private $exchange;

    /**
     * @var \AMQPConnection
     */
    private $connection;

    /**
     * @var \AMQPChannel
     */
    private $channel;

    public function __construct(AmqpConfiguration $configuration = null)
    {
        $this->configuration = $configuration ?: new AmqpConfiguration();
    }

    /**
     * @param $name
     * @return MessageQueue
     */
    public function getNamedQueue($name)
    {
        $exchange = $this->declareExchange();

        return $this->buildQueue($exchange, $name, $this->configuration->getDeadLetterExchange());
    }

    /**
     * @param \AMQPExchange $exchange
     * @param $queueName
     * @param string $errorExchangeName
     * @return \AMQPQueue
     */
    private function buildQueue(\AMQPExchange $exchange, $queueName, $errorExchangeName = '')
    {
        $queue = new \AMQPQueue($exchange->getChannel());
        $queue->setName($queueName);
        $queue->setFlags(AMQP_DURABLE);

        if (trim($errorExchangeName) != '') {
            $queue->setArgument('x-dead-letter-exchange', $errorExchangeName);
        }

        $queue->declareQueue();
        $queue->bind($exchange->getName(), $queueName);

        return new AmqpMessageQueue($exchange, $queue);
    }

    protected function declareExchange()
    {
        if ($this->exchange == null) {
            $this->exchange = new \AMQPExchange($this->getChannel());

            $this->exchange->setFlags(AMQP_DURABLE);

            if ($this->configuration->getExchange() != null) {
                $this->exchange->setType(AMQP_EX_TYPE_DIRECT);
                $this->exchange->setName($this->configuration->getExchange());
                $this->exchange->declareExchange();
            }
        }

        return $this->exchange;
    }


    /**
     * @return \AMQPConnection
     */
    public function getConnection()
    {
        if ($this->connection == null) {
            $this->connection = new \AMQPConnection($this->configuration->toConnectionArray());
            $this->connection->connect();
        }

        return $this->connection;
    }

    /**
     * @return \AMQPChannel
     */
    public function getChannel()
    {
        if ($this->channel == null) {
            $this->channel = new \AMQPChannel($this->getConnection());
        }

        return $this->channel;
    }
}