Cmobi\RabbitmqBundle\Queue\Queue

Complexity

Total Complexity

17

Size/Duplication

Total Lines	140
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	8

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	1
A	getChannel()	0	13	3
A	createQueue()	0	15	2
A	start()	0	18	3
A	getQueuebag()	0	4	1
A	setCallback()	0	4	1
A	getCallback()	0	4	1
A	getConnectionManager()	0	4	1
A	getConnection()	0	4	1
A	forceReconnect()	0	19	3

<?php

namespace Cmobi\RabbitmqBundle\Queue;

use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel;
use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnection;
use Cmobi\RabbitmqBundle\Connection\ConnectionManager;
use Cmobi\RabbitmqBundle\Connection\Exception\InvalidAMQPChannelException;
use Psr\Log\LoggerInterface;

class Queue implements QueueInterface
{
    private $connectionManager;
    private $connection;
    private $channel;
    private $queueBag;
    private $callback;
    private $logger;

    public function __construct(ConnectionManager $connectionManager, QueueBagInterface $queueBag, LoggerInterface $logger)
    {
        $this->connectionManager = $connectionManager;
        $this->connection = $this->getConnectionManager()->getConnection();
        $this->queueBag = $queueBag;
        $this->logger = $logger;
    }

    /**
     * @return CmobiAMQPChannel
     * @throws InvalidAMQPChannelException
     */
    protected function getChannel()
    {
        if ($this->channel instanceof CmobiAMQPChannel) {
            return $this->channel;
        }
        $this->channel = $this->getConnection()->channel();

        if (! $this->channel instanceof CmobiAMQPChannel) {
            throw new InvalidAMQPChannelException('Failed get AMQPChannel');
        }

        return $this->channel;
    }

    protected function createQueue()
    {
        $queueBag = $this->getQueuebag();

        $this->getChannel()->basic_qos(null, $queueBag->getBasicQos(), null);

null is of type null, but the function expects a boolean.

        if ($queueBag->getExchangeDeclare()) {
            $this->getChannel()->exchangeDeclare($queueBag->getExchangeDeclare());
            list ($queueName, , ) = $this->getChannel()->queueDeclare($queueBag->getQueueDeclare());
            $this->getChannel()->queue_bind($queueName, $queueBag->getExchange());
        } else {
            $this->getChannel()->queueDeclare($queueBag->getQueueDeclare());
        }
        $this->getChannel()->basicConsume($queueBag->getQueueConsume(), $this->getCallback()->toClosure());
    }

    /**
     * Declare and start queue in broker
     */
    public function start()
    {
        $this->createQueue();

        while(count($this->getChannel()->callbacks)) {
            try {
                $this->getChannel()->wait();
            } catch (\Exception $e) {
                $this->logger->error($e->getMessage());
                $this->forceReconnect();

                continue;
            }
        }
        $connection = $this->getChannel()->getConnection();
        $this->getChannel()->close();
        $connection->close();
    }

    /**
     * @return QueueBagInterface
     */
    public function getQueuebag()
    {
        return $this->queueBag;
    }

    /**
     * @param QueueCallbackInterface $callback
     */
    public function setCallback(QueueCallbackInterface $callback)
    {
        $this->callback = $callback;
    }

    /**
     * @return QueueCallbackInterface
     */
    public function getCallback()
    {
        return $this->callback;
    }

    /**
     * @return ConnectionManager
     */
    public function getConnectionManager()
    {
        return $this->connectionManager;
    }

    /**
     * @return CmobiAMQPConnection
     */
    public function getConnection()
    {
        return $this->connection;
    }

    /**
     * Retry connect to message broker until it can.
     */
    /**
     * @param CmobiAMQPConnection|null $connection
     * @return CmobiAMQPChannel
     */
    public function forceReconnect(CmobiAMQPConnection $connection = null)
    {
        do {
            try {
                $failed = false;
                $this->logger->warning('forceReconnect() - trying connect...');
                $this->connection = $this->getConnectionManager()->getConnection();
                $this->channel = $this->getConnection()->channel();
                $this->createQueue();
            } catch (\Exception $e) {
                $failed = true;
                sleep(3);
                $this->logger->error('forceReconnect() - ' . $e->getMessage());
            }
        } while ($failed);
        $this->logger->warning('forceReconnect() - connected!');

        return $this->channel;
    }
}