AMQPAL\Adapter\PhpAmqpLib\Exchange

Complexity

Total Complexity

14

Size/Duplication

Total Lines	189
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Test Coverage

Coverage

100%

Importance

Changes	3
Bugs	1	Features	0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	getOptions()	0	4	1
A	setOptions()	0	8	2
A	declareExchange()	0	15	1
A	delete()	0	6	1
A	bind()	0	10	2
A	unbind()	0	10	2
B	publish()	0	24	2
A	setChannel()	0	5	1
A	getChannel()	0	4	1
A	getConnection()	0	4	1

<?php

namespace AMQPAL\Adapter\PhpAmqpLib;

use PhpAmqpLib\Message\AMQPMessage;
use AMQPAL\Adapter\ExchangeInterface;
use AMQPAL\Options;
use AMQPAL\Exception as BaseException;

/**
 * Class Exchange
 *
 * @package AMQPAL\Adapter\PhpAmqpLib
 */
class Exchange implements ExchangeInterface
{
    /**
     * @var Channel
     */
    protected $channel;
    /**
     * @var Options\ExchangeOptions
     */
    protected $options;

    /**
     * @return Options\ExchangeOptions
     */
    public function getOptions()
    {
        return $this->options;
    }

    /**
     * @param Options\ExchangeOptions|\Traversable|array $exchangeOptions
     * @return $this
     * @throws BaseException\BadMethodCallException
     * @throws BaseException\InvalidArgumentException
     */
    public function setOptions($exchangeOptions)
    {
        if (!$exchangeOptions instanceof Options\ExchangeOptions) {
            $exchangeOptions = new Options\ExchangeOptions($exchangeOptions);
        }
        $this->options = $exchangeOptions;
        return $this;
    }

    /**
     * Declare a new exchange on the broker.
     *
     * @return $this
     */
    public function declareExchange()
    {
        $this->channel->getResource()->exchange_declare(
            $this->options->getName(),
            $this->options->getType(),
            $this->options->isPassive(),
            $this->options->isDurable(),
            $this->options->isAutoDelete(),
            $this->options->isInternal(),
            $this->options->isNoWait(),
            $this->options->getArguments()
        );

        return $this;
    }

    /**
     * Delete the exchange from the broker.
     *
     * @param bool   $ifUnused      Optional if the exchange should not be
     *                              deleted until no clients are connected to
     *                              it.
     * @param bool   $noWait        No wait for a reply
     *
     * @return $this
     */
    public function delete($ifUnused = false, $noWait = false)
    {
        $this->channel->getResource()->exchange_delete($this->options->getName(), $ifUnused, $noWait);

        return $this;
    }

    /**
     * Bind to another exchange.
     *
     * Bind an exchange to another exchange using the specified routing key.
     *
     * @param string $exchangeName Name of the exchange to bind.
     * @param string $routingKey   The routing key to use for binding.
     * @param bool   $noWait       No wait for a reply
     * @param array  $arguments    Additional binding arguments.
     *
     * @return $this
     */
    public function bind($exchangeName, $routingKey = null, $noWait = false, array $arguments = [])
    {
        if (null === $routingKey) {
            $routingKey = '';
        }
        $name = $this->options->getName();
        $this->channel->getResource()->exchange_bind($name, $exchangeName, $routingKey, $noWait, $arguments);

        return $this;
    }

    /**
     * Remove binding to another exchange.
     *
     * Remove a routing key binding on an another exchange from the given exchange.
     *
     * @param string $exchangeName Name of the exchange to bind.
     * @param string $routingKey   The routing key to use for binding.
     * @param array  $arguments    Additional binding arguments.
     *
     * @return $this
     */
    public function unbind($exchangeName, $routingKey = null, array $arguments = [])
    {
        if (null === $routingKey) {
            $routingKey = '';
        }
        $name = $this->options->getName();
        $this->channel->getResource()->exchange_unbind($name, $exchangeName, $routingKey, $arguments);

        return $this;
    }

    /**
     * Publish a message to an exchange.
     *
     * Publish a message to the exchange represented by the Exchange object.
     *
     * @param string $message      The message to publish.
     * @param string $routingKey   The optional routing key to which to
     *                             publish to.
     * @param bool   $mandatory    Mandatory
     * @param bool   $immediate    Immediate
     * @param array  $attributes   One of content_type, content_encoding,
     *                             message_id, user_id, app_id, delivery_mode,
     *                             priority, timestamp, expiration, type
     *                             or reply_to, headers.
     *
     * @return $this
     */
    public function publish(
        $message,
        $routingKey = null,
        $mandatory = false,
        $immediate = false,
        array $attributes = []
    ) {
        if (null === $routingKey) {
            $routingKey = '';
        }
        $options = $this->options;

        $AMQPMessage = new AMQPMessage($message, $attributes);

        $this->channel->getResource()->basic_publish(
            $AMQPMessage,
            $options->getName(),
            $routingKey,
            $mandatory,
            $immediate
        );

        return $this;
    }

    /**
     * @param Channel $channel
     * @return $this
     */
    public function setChannel(Channel $channel)
    {
        $this->channel = $channel;
        return $this;
    }

    /**
     * Get the Channel object in use
     *
     * @return Channel
     */
    public function getChannel()
    {
        return $this->channel;
    }

    /**
     * Get the Connection object in use
     *
     * @return Connection
     */
    public function getConnection()
    {
        return $this->channel->getConnection();
    }
}