Cmobi\RabbitmqBundle\Rpc\RpcServer

Complexity

Total Complexity

18

Size/Duplication

Total Lines	155
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	9

Importance

Changes	4
Bugs	1	Features	2

9 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	11	3
A	pushMessage()	0	17	1
B	run()	0	46	6
A	buildChannel()	0	14	3
A	getConnection()	0	4	1
A	getChannel()	0	4	1
A	getContainer()	0	4	1
A	getLogger()	0	4	1
A	setLogger()	0	4	1

<?php

namespace Cmobi\RabbitmqBundle\Rpc;

use Cmobi\RabbitmqBundle\Rpc\Exception\InvalidRpcServerClassException;
use Cmobi\RabbitmqBundle\Rpc\Exception\NotFoundRpcServiceException;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exception\AMQPRuntimeException;
use Psr\Log\LoggerInterface;
use Symfony\Component\DependencyInjection\ContainerAwareTrait;

class RpcServer
{
    use ContainerAwareTrait;

    private $rpcServices;
    private $connection;
    private $logger;
    private $channel;

    public function __construct(array $rpcServices, AMQPStreamConnection $connection = null)
    {
        if (!$rpcServices) {

The expression $rpcServices of type array is implicitly converted to a boolean; are you sure this is intended? If so, consider using empty($expr) instead to make it clear that you intend to check for an array without elements.

            throw new NotFoundRpcServiceException('no rpc services found.');
        }
        $this->rpcServices = $rpcServices;

        if (!is_null($connection)) {
            $this->connection = $connection;
        }
    }

    /**
     * @param $queue
     * @param RpcServiceInterface $serviceCallback
     * @param bool|false $passive
     * @param bool|false $durable
     * @param bool|false $exclusive
     * @param bool|true $auto_delete
     * @param bool|false $nowait
     * @param null $arguments
     * @param null $ticket
     */
    public function pushMessage(
        $queue,
        RpcServiceInterface $serviceCallback,
        $passive = false,
        $durable = false,
        $exclusive = false,
        $auto_delete = true,
        $nowait = false,
        $arguments = null,
        $ticket = null
    )
    {
        $this->getChannel()->queue_declare(
            $queue, $passive, $durable, $exclusive, $auto_delete, $nowait, $arguments, $ticket
        );
        $this->getChannel()->basic_consume($queue, '', false, false, $exclusive, $nowait, $serviceCallback->createCallback());
    }

    public function run()
    {
        foreach ($this->rpcServices as $serviceName) {
            $service = $this->getContainer()->get($serviceName);

            if (!$service instanceof RpcServiceInterface) {
                throw new InvalidRpcServerClassException(
                    'Failed start RpcServer: %s is not instance of RpcServiceInterface'. $serviceName
                );
            }
            list(
                $name, $passive, $durable, $exclusive, $auto_delete, $nowait, $arguments, $ticket

The assignment to $name is unused. Consider omitting it like so list($first,,$third).

                ) = array_values($service->getQueueOptions());
            $this->pushMessage(
                $service->getQueueName(),
                $service,
                $passive,
                $durable,
                $exclusive,
                $auto_delete,
                $nowait,
                $arguments,
                $ticket
            );
        }

        while(count($this->getChannel()->callbacks)) {
            try {
                $this->getChannel()->wait();
            } catch (AMQPRuntimeException $e) {

                if ($this->logger instanceof LoggerInterface) {
                    $this->logger->error(
                        sprintf(
                            'Failed process queue with error: %s',
                            $e->getMessage()
                        )
                    );
                }
                continue;
            }
        }

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

    /**
     * @throws \Cmobi\RabbitmqBundle\Amqp\Exception\NotFoundAMQPConnectionFactoryException
     */
    public function buildChannel()
    {
        if (!$this->connection instanceof AMQPStreamConnection) {
            $connectionManager = $this->getContainer()->get('cmobi_rabbitmq.connection.manager');
            $this->connection = $connectionManager->getConnection();
        }
        $this->channel = $this->connection->channel();
        $qos = 1;

        if ($this->getContainer()->hasParameter('cmobi_rabbitmq.basic_qos')) {
            $qos = $this->getContainer()->getParameter('cmobi_rabbitmq.basic_qos');
        }
        $this->getChannel()->basic_qos(null, $qos, null);

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

    }

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

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

    /**
     * @return \Symfony\Component\DependencyInjection\ContainerInterface
     */
    protected function getContainer()
    {
        return $this->container;
    }

    /**
     * @return LoggerInterface
     */
    public function getLogger()
    {
        return $this->logger;
    }

    /**
     * @param LoggerInterface $logger
     */
    public function setLogger(LoggerInterface $logger)
    {
        $this->logger = $logger;
    }
}