Puzzle\AMQP\Clients\Pecl

Complexity

Total Complexity

16

Size/Duplication

Total Lines	137
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

0%

Importance

Changes

0

9 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	7	1
B	ensureIsConnected()	0	24	3
A	publish()	0	15	2
A	logMessage()	0	9	1
A	sendMessage()	0	22	2
A	computeMessageFlag()	0	12	3
A	getExchange()	0	16	2
A	updateMessageAttributes()	0	5	1
A	getQueue()	0	9	1

<?php

namespace Puzzle\AMQP\Clients;

use Puzzle\Configuration;
use Puzzle\PrefixedConfiguration;

use Puzzle\AMQP\Client;
use Psr\Log\LoggerAwareTrait;
use Psr\Log\NullLogger;
use Puzzle\AMQP\WritableMessage;

class Pecl implements Client
{
    use LoggerAwareTrait;

    private
        $applicationId,

The visibility should be declared for property $applicationId.

        $configuration,
        $channel;

    public function __construct(Configuration $configuration)
    {
        $this->applicationId = $configuration->read('app/id', 'Unknown application');
        $this->configuration = $configuration;
        $this->channel = null;
        $this->logger = new NullLogger();
    }

    private function ensureIsConnected()
    {
        if(! $this->channel instanceof \AMQPChannel)
        {
            $configuration = new PrefixedConfiguration($this->configuration, 'amqp/broker');

            // Create a connection
            $connection = new \AMQPConnection();
            $connection->setHost($configuration->readRequired('host'));
            $connection->setLogin($configuration->readRequired('login'));
            $connection->setPassword($configuration->readRequired('password'));

            $vhost = $configuration->read('vhost', null);
            if($vhost !== null)
            {
                $connection->setVhost($vhost);
            }

            $connection->connect();

            // Create a channel
            $this->channel = new \AMQPChannel($connection);
        }
    }

    public function publish($exchangeName, WritableMessage $message)
    {
        try
        {
            $ex = $this->getExchange($exchangeName);
        }
        catch(\Exception $e)
        {
            $this->logMessage($exchangeName, $message);

            return false;
        }

        return $this->sendMessage($ex, $message);
    }

    private function logMessage($exchangeName, WritableMessage $message)
    {
        $log = json_encode(array(
            'exchange' => $exchangeName,
            'message' => (string) $message,
        ));

        $this->logger->error($log, ['This message was involved by an error (it was sent ... or not. Please check other logs)']);
    }

    private function sendMessage(\AMQPExchange $ex, WritableMessage $message)
    {
        try
        {
            $this->updateMessageAttributes($message);

            $ex->publish(
                $message->getBodyInTransportFormat(),
                $message->getRoutingKey(),
                $this->computeMessageFlag(),
                $message->packAttributes()
            );
        }
        catch(\Exception $e)
        {
            $this->logMessage($ex->getName(), $message);

            return false;
        }

        return true;
    }
    
    private function computeMessageFlag()
    {
        $flag = AMQP_NOPARAM;
        $disallowSilentDropping = $this->configuration->read('amqp/global/disallowSilentDropping', false);
        
        if($disallowSilentDropping === true || $message->canBeDroppedSilently() === false)

The variable $message does not exist. Did you forget to declare it?

        {
            $flag = AMQP_MANDATORY;
        }
        
        return $flag;
    }

    public function getExchange($exchangeName = null, $type = AMQP_EX_TYPE_TOPIC)
    {
        $this->ensureIsConnected();

        $ex = new \AMQPExchange($this->channel);

        if(!empty($exchangeName))
        {
            $ex->setName($exchangeName);
            $ex->setType($type);
            $ex->setFlags(AMQP_PASSIVE);
            $ex->declareExchange();
        }

        return $ex;
    }

    private function updateMessageAttributes(WritableMessage $message)
    {
        $message->setAttribute('app_id', $this->applicationId);
        $message->addHeader('routing_key', $message->getRoutingKey());
    }

    public function getQueue($queueName)
    {
        $this->ensureIsConnected();

        $queue = new \AMQPQueue($this->channel);
        $queue->setName($queueName);

        return $queue;
    }
}