OldSound\RabbitMqBundle\RabbitMq\Producer

Complexity

Total Complexity

13

Size/Duplication

Total Lines	114
Duplicated Lines	0 %

Test Coverage

Coverage

0%

Importance

Changes	13
Bugs	0	Features	0

9 Methods

Rating	Name	Duplication	Size	Complexity
A	reconnect()	0	4	1
A	__construct()	0	5	1
A	publish()	0	25	4
A	initializeProducer()	0	13	2
A	setDeliveryMode()	0	5	1
A	getBasicProperties()	0	3	1
A	setDefaultRoutingKey()	0	5	1
A	setContentType()	0	5	1
A	setConfirmationTimeout()	0	3	1

<?php

namespace OldSound\RabbitMqBundle\RabbitMq;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AbstractConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

/**
 * Producer, that publishes AMQP Messages
 */
class Producer extends BaseAmqp implements ProducerInterface
{
    public const DEFAULT_CONTENT_TYPE = 'text/plain';
    protected $contentType = Producer::DEFAULT_CONTENT_TYPE;
    protected $deliveryMode = 2;
    protected $defaultRoutingKey = '';
    protected $acknowledged = true;
    protected $confirmationTimeout = 0;
    protected $confirmSelect = false;

    public function setContentType($contentType)
    {
        $this->contentType = $contentType;

        return $this;
    }

    public function setDeliveryMode($deliveryMode)
    {
        $this->deliveryMode = $deliveryMode;

        return $this;
    }

    public function setDefaultRoutingKey($defaultRoutingKey)
    {
        $this->defaultRoutingKey = $defaultRoutingKey;

        return $this;
    }

    protected function getBasicProperties()
    {
        return array('content_type' => $this->contentType, 'delivery_mode' => $this->deliveryMode);
    }

    /**
     * @param null $confirmationTimeout

Are you sure the doc-type for parameter $confirmationTimeout is correct as it would always require null to be passed?

     */
    public function setConfirmationTimeout($confirmationTimeout): void
    {
        $this->confirmationTimeout = intval($confirmationTimeout);
    }

    /**
     * @param AbstractConnection $conn
     * @param AMQPChannel|null $ch
     * @param null $consumerTag

Are you sure the doc-type for parameter $consumerTag is correct as it would always require null to be passed?

     * @param bool $confirmSelect
     */
    public function __construct(AbstractConnection $conn, AMQPChannel $ch = null, $consumerTag = null, bool $confirmSelect = false)
    {
        parent::__construct($conn, $ch, $consumerTag);
        $this->confirmSelect = $confirmSelect;
        $this->initializeProducer();
    }

    /**
     * Publishes the message and merges additional properties with basic properties
     *
     * @param string $msgBody
     * @param string $routingKey
     * @param array $additionalProperties
     * @param array|null $headers
     * @return bool
     */
    public function publish($msgBody, $routingKey = '', $additionalProperties = array(), array $headers = null)
    {
        if ($this->autoSetupFabric) {
            $this->setupFabric();
        }

        $msg = new AMQPMessage((string) $msgBody, array_merge($this->getBasicProperties(), $additionalProperties));

        if (!empty($headers)) {
            $headersTable = new AMQPTable($headers);
            $msg->set('application_headers', $headersTable);
        }

        $real_routingKey = !empty($routingKey) ? $routingKey : $this->defaultRoutingKey;
        $this->getChannel()->basic_publish($msg, $this->exchangeOptions['name'], (string)$real_routingKey);
        $this->getChannel()->wait_for_pending_acks($this->confirmationTimeout);
        $this->logger->debug('AMQP message published', array(
            'amqp' => array(
                'body' => $msgBody,
                'routingkeys' => $routingKey,
                'properties' => $additionalProperties,
                'headers' => $headers
            )
        ));
        return $this->acknowledged;
    }

    public function reconnect()
    {
        parent::reconnect();
        $this->initializeProducer();
    }

    /**
     */
    protected function initializeProducer(): void
    {
        if ($this->confirmSelect) {
            $this->getChannel()->confirm_select();
            $this->getChannel()->set_ack_handler(
                function (AMQPMessage $message) {

The parameter $message is not used and could be removed.

                    $this->acknowledged = true;
                }
            );

            $this->getChannel()->set_nack_handler(
                function (AMQPMessage $message) {

The parameter $message is not used and could be removed.

                    $this->acknowledged = false;
                }
            );
        }
    }
}