OrderManagement\RabbitMQ\Consumer

Complexity

Total Complexity

5

Size/Duplication

Total Lines	65
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	addCallback()	0	3	1
A	__construct()	0	3	1
A	work()	0	33	3

<?php
/**
 * This file is part of order_management
 * User: Sinan TURGUT <mail@sinanturgut.com.tr>
 * Date: 24.06.2019
 * php version 7.2
 *
 * @category Assessment
 * @package  OrderManagement
 * @author   Sinan TURGUT <mail@sinanturgut.com.tr>
 * @license  See LICENSE file
 * @link     https://dev.sinanturgut.com.tr
 */

namespace OrderManagement\RabbitMQ;

/**
 * Class Consumer
 * @package OrderManagement\RabbitMQ
 */
class Consumer
{
    /**
     * @var AbstractQueue
     */
    protected $queue;

    /**
     * @var array
     */
    protected $callbacks = [];

    /**
     * @param QueueInterface $queue
     */
    public function __construct(QueueInterface $queue)
    {
        $this->queue = $queue;

$queue is of type OrderManagement\RabbitMQ\QueueInterface, but the property $queue was declared to be of type OrderManagement\RabbitMQ\AbstractQueue. Are you sure that you always receive this specific sub-class here, or does it make sense to add an instanceof check?

    }

    /**
     * @param $callable
     */
    public function addCallback($callable)
    {
        $this->callbacks[] = $callable;
    }

    /**
     * @param callable $stream
     * @throws \ErrorException
     */
    public function work(callable $stream)
    {
        $connection = new Connection();
        $channel = $connection->channel();

        $channel->queue_declare(
            $this->queue->getName(),
            $this->queue->isPassive(),
            $this->queue->isDurable(),
            $this->queue->isExclusive(),
            $this->queue->isAutoDelete()
        );

        $stream('[*] Waiting for messages. To exit press CTRL+C');

        foreach ($this->callbacks as $callback) {
            $channel->basic_consume(
                $this->queue->getName(),
                '',
                false,
                true,
                false,
                false,
                $callback
            );
        }

        while (count($channel->callbacks)) {
            $channel->wait();
        }

        $channel->close();
        $connection->close();
    }
}