Cmobi\RabbitmqBundle\Transport\Subscriber\Publisher

Complexity

Total Complexity

9

Size/Duplication

Total Lines	98
Duplicated Lines	12.24 %

Coupling/Cohesion

Components	1
Dependencies	5

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	13	1
A	publish()	0	11	1
A	refreshChannel()	12	12	2
A	getChannel()	0	4	1
A	getQueueName()	0	4	1
A	getFromName()	0	4	1
A	getExchange()	0	4	1
A	getExchangeType()	0	4	1

<?php

namespace Cmobi\RabbitmqBundle\Transport\Subscriber;

use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel;
use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnectionInterface;
use Cmobi\RabbitmqBundle\Connection\ConnectionManager;
use Cmobi\RabbitmqBundle\Queue\CmobiAMQPMessage;
use Cmobi\RabbitmqBundle\Queue\QueueProducerInterface;

class Publisher implements QueueProducerInterface
{
    private $connectionManager;
    private $channel;
    private $fromName;
    private $queueName;
    private $exchange;
    private $exchangeType;

    public function __construct(
        $exchange,
        $exchangeType = ExchangeType::FANOUT,
        ConnectionManager $manager,
        $fromName = '',
        $queueName = ''
    ) {
        $this->exchange = $exchange;
        $this->exchangeType = $exchangeType;
        $this->queueName = $queueName;
        $this->fromName = $fromName;
        $this->connectionManager = $manager;
    }

    /**
     * @param $data
     * @param $expire
     * @param $priority
     */
    public function publish($data, $expire = self::DEFAULT_TTL, $priority = self::PRIORITY_LOW)
    {
        $this->refreshChannel();
        $queueBag = new SubscriberQueueBag($this->getExchange(), $this->getExchangeType(), $this->getQueueName());
        $this->getChannel()->exchangeDeclare($queueBag->getExchangeDeclare());

It seems like $queueBag->getExchangeDeclare() targeting Cmobi\RabbitmqBundle\Transport\Subscriber\SubscriberQueueBag::getExchangeDeclare() can also be of type false; however, Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel::exchangeDeclare() does only seem to accept array, did you maybe forget to handle an error condition?

        $msg = new CmobiAMQPMessage((string) $data);
        $this->getChannel()->basic_publish($msg, $queueBag->getExchange());

        $this->getChannel()->close();
        $this->connectionManager->getConnection()->close();
    }

    /**
     * @return CmobiAMQPChannel
     */
    public function refreshChannel()

This method seems to be duplicated in your project.

    {
        /** @var CmobiAMQPConnectionInterface $connection */
        $connection = $this->connectionManager->getConnection();

        if (!$connection->isConnected()) {
            $connection->reconnect();
        }
        $this->channel = $connection->channel();

        return $this->channel;
    }

    /**
     * @return CmobiAMQPChannel
     */
    public function getChannel()
    {
        return $this->channel;
    }

    /**
     * @return string
     */
    public function getQueueName()
    {
        return $this->queueName;
    }

    /**
     * Return caller name.
     *
     * @return string
     */
    public function getFromName()
    {
        return $this->fromName;
    }

    /**
     * @return string
     */
    public function getExchange()
    {
        return $this->exchange;
    }

    /**
     * @return string
     */
    public function getExchangeType()
    {
        return $this->exchangeType;
    }
}