Queue::setCallback() - Code Metrics - contamobi/CmobiRabbitmqBundle - Measure and Improve Code Quality continuously with Scrutinizer

Queue::setCallback() A
last analyzed 2017-05-30 00:09 UTC

↳ Parent: Queue

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	4
rs	10
c	0
b	0
f	0
cc	1
eloc	2
nc	1
nop	1

<?php

namespace Cmobi\RabbitmqBundle\Queue;

use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel;
use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnection;
use Cmobi\RabbitmqBundle\Connection\ConnectionManager;
use Cmobi\RabbitmqBundle\Connection\Exception\InvalidAMQPChannelException;
use Psr\Log\LoggerInterface;

class Queue implements QueueInterface
{
    private $connectionManager;
    private $connection;
    private $connectionName;
    private $channel;
    private $queueBag;
    private $errOutput;
    private $logOutput;
    private $callback;

    public function __construct(
        ConnectionManager $connectionManager,
        QueueBagInterface $queueBag,
        $connectionName = 'default',
        QueueCallbackInterface $callback = null
    )
    {
        $this->connectionManager = $connectionManager;
        $this->connectionName = $connectionName;
        $this->connection = $this->getConnectionManager()->getConnection($connectionName);
        $this->queueBag = $queueBag;
        $this->errOutput = fopen('php://stderr', 'a+');
        $this->logOutput = fopen('php://stdout', 'a+');
        $this->callback = $callback;
    }

    /**
     * @return CmobiAMQPChannel
     *
     * @throws InvalidAMQPChannelException
     */
    protected function getChannel()
    {
        if ($this->channel instanceof CmobiAMQPChannel) {
            return $this->channel;
        }
        $this->channel = $this->getConnection()->channel();

        if (!$this->channel instanceof CmobiAMQPChannel) {
            throw new InvalidAMQPChannelException('Failed get AMQPChannel');
        }

        return $this->channel;
    }

    protected function createQueue()
    {
        $queueBag = $this->getQueuebag();

        $this->getChannel()->basic_qos(null, $queueBag->getBasicQos(), null);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);

        if ($queueBag->getExchangeDeclare()) {
            $this->getChannel()->exchangeDeclare($queueBag->getExchangeDeclare());
            list($queueName) = $this->getChannel()->queueDeclare($queueBag->getQueueDeclare());
            $this->getChannel()->queue_bind($queueName, $queueBag->getExchange());
        } else {
            $this->getChannel()->queueDeclare($queueBag->getQueueDeclare());
        }
        $this->getChannel()->basicConsume($queueBag->getQueueConsume(), $this->getCallback()->toClosure());
    }

    /**
     * Declare and start queue in broker.
     */
    public function start()
    {
        $this->createQueue();

        while (count($this->getChannel()->callbacks)) {
            try {
                $this->getChannel()->wait();
            } catch (\Exception $e) {
                fwrite($this->errOutput, $e->getMessage());
                $this->forceReconnect();

                continue;
            }
        }
        $connection = $this->getChannel()->getConnection();
        $this->getChannel()->close();
        $connection->close();
    }

    /**
     * @return QueueBagInterface
     */
    public function getQueuebag()
    {
        return $this->queueBag;
    }

    /**
     * @param QueueCallbackInterface $callback
     */
    public function setCallback(QueueCallbackInterface $callback)
    {
        $this->callback = $callback;
    }

    /**
     * @return QueueCallbackInterface
     */
    public function getCallback()
    {
        return $this->callback;
    }

    /**
     * @return ConnectionManager
     */
    public function getConnectionManager()
    {
        return $this->connectionManager;
    }

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

    /**
     * Retry connect to message broker until it can.
     */
    /**
     * @param CmobiAMQPConnection|null $connection
     *
     * @return CmobiAMQPChannel
     */
    public function forceReconnect(CmobiAMQPConnection $connection = null)
    {
        do {
            try {
                $failed = false;
                fwrite($this->logOutput, 'start Queue::forceReconnect() - trying connect...' . PHP_EOL);
                $this->connection = $this->getConnectionManager()->getConnection($this->connectionName);
                $this->channel = $this->getConnection()->channel();
                $this->createQueue();
            } catch (\Exception $e) {
                $failed = true;
                sleep(3);
                fwrite($this->errOutput, 'failed Queue::forceReconnect() - ' . $e->getMessage() . PHP_EOL);
            }
        } while ($failed);
        fwrite($this->logOutput, 'Queue::forceReconnect() - connected!' . PHP_EOL);

        return $this->channel;
    }
}


1			<?php
2
3			namespace Cmobi\RabbitmqBundle\Queue;
4
5			use Cmobi\RabbitmqBundle\Connection\CmobiAMQPChannel;
6			use Cmobi\RabbitmqBundle\Connection\CmobiAMQPConnection;
7			use Cmobi\RabbitmqBundle\Connection\ConnectionManager;
8			use Cmobi\RabbitmqBundle\Connection\Exception\InvalidAMQPChannelException;
9			use Psr\Log\LoggerInterface;
10
11			class Queue implements QueueInterface
12			{
13			private $connectionManager;
14			private $connection;
15			private $connectionName;
16			private $channel;
17			private $queueBag;
18			private $errOutput;
19			private $logOutput;
20			private $callback;
21
22			public function __construct(
23			ConnectionManager $connectionManager,
24			QueueBagInterface $queueBag,
25			$connectionName = 'default',
26			QueueCallbackInterface $callback = null
27			)
28			{
29			$this->connectionManager = $connectionManager;
30			$this->connectionName = $connectionName;
31			$this->connection = $this->getConnectionManager()->getConnection($connectionName);
32			$this->queueBag = $queueBag;
33			$this->errOutput = fopen('php://stderr', 'a+');
34			$this->logOutput = fopen('php://stdout', 'a+');
35			$this->callback = $callback;
36			}
37
38			/**
39			* @return CmobiAMQPChannel
40			*
41			* @throws InvalidAMQPChannelException
42			*/
43			protected function getChannel()
44			{
45			if ($this->channel instanceof CmobiAMQPChannel) {
46			return $this->channel;
47			}
48			$this->channel = $this->getConnection()->channel();
49
50			if (!$this->channel instanceof CmobiAMQPChannel) {
51			throw new InvalidAMQPChannelException('Failed get AMQPChannel');
52			}
53
54			return $this->channel;
55			}
56
57			protected function createQueue()
58			{
59			$queueBag = $this->getQueuebag();
60
61			$this->getChannel()->basic_qos(null, $queueBag->getBasicQos(), null);
			0 ignored issues – show Documentation introduced 2016-12-28 18:51 UTC by Report Bug Copy Issue Report `null` is of type `null`, but the function expects a `boolean`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
62
63			if ($queueBag->getExchangeDeclare()) {
64			$this->getChannel()->exchangeDeclare($queueBag->getExchangeDeclare());
65			list($queueName) = $this->getChannel()->queueDeclare($queueBag->getQueueDeclare());
66			$this->getChannel()->queue_bind($queueName, $queueBag->getExchange());
67			} else {
68			$this->getChannel()->queueDeclare($queueBag->getQueueDeclare());
69			}
70			$this->getChannel()->basicConsume($queueBag->getQueueConsume(), $this->getCallback()->toClosure());
71			}
72
73			/**
74			* Declare and start queue in broker.
75			*/
76			public function start()
77			{
78			$this->createQueue();
79
80			while (count($this->getChannel()->callbacks)) {
81			try {
82			$this->getChannel()->wait();
83			} catch (\Exception $e) {
84			fwrite($this->errOutput, $e->getMessage());
85			$this->forceReconnect();
86
87			continue;
88			}
89			}
90			$connection = $this->getChannel()->getConnection();
91			$this->getChannel()->close();
92			$connection->close();
93			}
94
95			/**
96			* @return QueueBagInterface
97			*/
98			public function getQueuebag()
99			{
100			return $this->queueBag;
101			}
102
103			/**
104			* @param QueueCallbackInterface $callback
105			*/
106			public function setCallback(QueueCallbackInterface $callback)
107			{
108			$this->callback = $callback;
109			}
110
111			/**
112			* @return QueueCallbackInterface
113			*/
114			public function getCallback()
115			{
116			return $this->callback;
117			}
118
119			/**
120			* @return ConnectionManager
121			*/
122			public function getConnectionManager()
123			{
124			return $this->connectionManager;
125			}
126
127			/**
128			* @return CmobiAMQPConnection
129			*/
130			public function getConnection()
131			{
132			return $this->connection;
133			}
134
135			/**
136			* Retry connect to message broker until it can.
137			*/
138			/**
139			* @param CmobiAMQPConnection\|null $connection
140			*
141			* @return CmobiAMQPChannel
142			*/
143			public function forceReconnect(CmobiAMQPConnection $connection = null)
144			{
145			do {
146			try {
147			$failed = false;
148			fwrite($this->logOutput, 'start Queue::forceReconnect() - trying connect...' . PHP_EOL);
149			$this->connection = $this->getConnectionManager()->getConnection($this->connectionName);
150			$this->channel = $this->getConnection()->channel();
151			$this->createQueue();
152			} catch (\Exception $e) {
153			$failed = true;
154			sleep(3);
155			fwrite($this->errOutput, 'failed Queue::forceReconnect() - ' . $e->getMessage() . PHP_EOL);
156			}
157			} while ($failed);
158			fwrite($this->logOutput, 'Queue::forceReconnect() - connected!' . PHP_EOL);
159
160			return $this->channel;
161			}
162			}
163

contamobi / CmobiRabbitmqBundle

Queue::setCallback() A last analyzed 2017-05-30 00:09 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Queue::setCallback() A
last analyzed 2017-05-30 00:09 UTC