Queue::consume() - Code Metrics - Inspection of "Codigo simplificado" - WillRy/rabbitrun - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 1f8a6a...d3094e )

by William

created 2023-09-23 22:25 UTC

Queue::consume() A

↳ Parent: Queue

Complexity

Conditions	5
Paths	1

Size

Total Lines	56
Code Lines	30

Duplication

Lines	0
Ratio	0 %

Importance

Changes	12
Bugs	0	Features	0

Metric	Value
cc	5
eloc	30
c	12
b	0
f	0
nc	1
nop	2
dl	0
loc	56
rs	9.1288

How to fix Long Method

<?php

declare(ticks=1);

namespace WillRy\RabbitRun\Queue;


use Exception;
use PhpAmqpLib\Message\AMQPMessage;
use WillRy\RabbitRun\Base;

class Queue extends Base
{
    /** @var string nome da fila */
    protected $queueName;

    /** @var string nome da exchange */
    protected $exchangeName;


    /** @var string nome do consumer */
    public $consumerName;

    protected \Closure $onReceiveCallback;

    protected \Closure $onExecutingCallback;

    protected \Closure $onErrorCallback;

    protected \Closure $onCheckStatusCallback;

    public function __construct()
    {
        parent::__construct();
    }

    public function shutdown($signal)
    {
        parent::shutdown($signal);
    }

    /**
     * Inicializa a fila e exchange, vinculano os 2
     * @param string $name
     * @return $this
     */
    public function createQueue(string $name)
    {
        $this->getConnection();

        $this->queueName = "{$name}";

        $this->exchangeName = "{$name}_exchange";

        $this->exchange($this->exchangeName);

        $this->queue($name);

        $this->bind($name, $this->exchangeName);

        return $this;
    }


    /**
     * Publica mensagem
     *
     * @param array $job
     * @return array
     * @throws Exception
     */
    public function publish(
        string $queueName,
        array  $job
    ) {
        try {
            $this->createQueue($queueName);

            $this->getConnection();

            $payload = [
                "payload" => $job,
                'queue' => $this->queueName,
            ];

            $json = json_encode($payload);

            $message = new AMQPMessage(
                $json,
                array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
            );

            $this->channel->basic_publish(
                $message,
                $this->exchangeName
            );

            return $payload;
        } catch (Exception $e) {
            throw $e;
        }
    }


    /**
     * Loop de consumo de mensagem
     *
     * @param int $sleepSeconds
     * @throws Exception
     */
    public function consume(
        string $queueName,
        int    $sleepSeconds = 3
    ) {
        $this->validateExecuteCallback();
        

        $this->loopConnection(function () use ($sleepSeconds, $queueName) {

            $this->createQueue($queueName);

            $this->channel->basic_qos(null, 1, null);

            $this->channel->basic_consume(
                $this->queueName,
                $this->randomConsumer(),
                false,
                false,
                false,
                false,
                function (AMQPMessage $message) {
                    pcntl_sigprocmask(SIG_BLOCK, [SIGTERM, SIGINT]);

                    $incomeData = json_decode($message->getBody(), true);

                    //se o status for negativo, não executa o consumo
                    $statusBoolean = $this->executeStatusCallback($message);

                    if (!$statusBoolean) {
                        return false;
                    }

                    $receiveBoolean = $this->executeReceiveCallback($message, $incomeData);

                    if (!$receiveBoolean) {
                        return false;
                    }

                    try {
                        $this->executeMessage($message, $incomeData);
                    } catch (Exception $e) {
                        $message->nack(true);
                        $this->executeErrorCallback($e, $incomeData);
                    }

                    pcntl_sigprocmask(SIG_UNBLOCK, [SIGTERM, SIGINT]);
                }
            );

            // Loop as long as the channel has callbacks registered
            while ($this->channel->is_open()) {
                $this->channel->wait(null, false);
                sleep($sleepSeconds);

                // Despachar eventos de "finalizar" script
                pcntl_signal_dispatch();
            }
        });
    }

    public function onCheckStatus(\Closure $callback)
    {
        $this->onCheckStatusCallback = $callback;
    }

    public function onReceive(\Closure $callback)
    {
        $this->onReceiveCallback = $callback;
    }

    public function onExecuting(\Closure $callback)
    {
        $this->onExecutingCallback = $callback;
    }

    public function onError(\Closure $callback)
    {
        $this->onErrorCallback = $callback;
    }

    public function executeStatusCallback(AMQPMessage $message)
    {
        if (empty($this->onCheckStatusCallback)) {
            return true;
        }

        $checkStatusCallback = $this->onCheckStatusCallback;
        $statusBoolean = $checkStatusCallback();

        if (!$statusBoolean && isset($statusBoolean)) {
            print_r("[WORKER STOPPED]" . PHP_EOL);
            $message->nack(true);
            pcntl_sigprocmask(SIG_UNBLOCK, [SIGTERM, SIGINT]);
            return false;
        }

        return true;
    }

    public function executeReceiveCallback(AMQPMessage $message, $incomeData)
    {
        if (empty($this->onReceiveCallback)) {
            return true;
        }

        $receiveCallback = $this->onReceiveCallback;
        $statusBoolean = $receiveCallback($incomeData);
        if (!$statusBoolean && isset($statusBoolean)) {
            print_r("[TASK IGNORED BY ON RECEIVE RETURN]" . PHP_EOL);
            $message->nack();
            pcntl_sigprocmask(SIG_UNBLOCK, [SIGTERM, SIGINT]);
            return false;
        }


        return true;
    }

    public function executeMessage(AMQPMessage $message, $incomeData)
    {
        $onExecutingCallback = $this->onExecutingCallback;
        $onExecutingCallback($message, $incomeData);
    }
    

    public function executeErrorCallback(\Exception $e, $incomeData)
    {
        if (empty($this->onErrorCallback)) {
            return false;
        }

        $errorCallback = $this->onErrorCallback;
        $errorCallback($e, $incomeData);
        return true;
    }

    public function validateExecuteCallback()
    {
        if (!empty($this->onExecutingCallback)) {
            return true;
        }

        throw new \Exception("Define a onExecuting callback");
    }
}


1			<?php
2
3			declare(ticks=1);
4
5			namespace WillRy\RabbitRun\Queue;
6
7
8			use Exception;
9			use PhpAmqpLib\Message\AMQPMessage;
10			use WillRy\RabbitRun\Base;
11
12			class Queue extends Base
13			{
14			/** @var string nome da fila */
15			protected $queueName;
16
17			/** @var string nome da exchange */
18			protected $exchangeName;
19
20
21			/** @var string nome do consumer */
22			public $consumerName;
23
24			protected \Closure $onReceiveCallback;
25
26			protected \Closure $onExecutingCallback;
27
28			protected \Closure $onErrorCallback;
29
30			protected \Closure $onCheckStatusCallback;
31
32			public function __construct()
33			{
34			parent::__construct();
35			}
36
37			public function shutdown($signal)
38			{
39			parent::shutdown($signal);
40			}
41
42			/**
43			* Inicializa a fila e exchange, vinculano os 2
44			* @param string $name
45			* @return $this
46			*/
47			public function createQueue(string $name)
48			{
49			$this->getConnection();
50
51			$this->queueName = "{$name}";
52
53			$this->exchangeName = "{$name}_exchange";
54
55			$this->exchange($this->exchangeName);
56
57			$this->queue($name);
58
59			$this->bind($name, $this->exchangeName);
60
61			return $this;
62			}
63
64
65			/**
66			* Publica mensagem
67			*
68			* @param array $job
69			* @return array
70			* @throws Exception
71			*/
72			public function publish(
73			string $queueName,
74			array $job
75			) {
76			try {
77			$this->createQueue($queueName);
78
79			$this->getConnection();
80
81			$payload = [
82			"payload" => $job,
83			'queue' => $this->queueName,
84			];
85
86			$json = json_encode($payload);
87
88			$message = new AMQPMessage(
89			$json,
90			array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
91			);
92
93			$this->channel->basic_publish(
94			$message,
95			$this->exchangeName
96			);
97
98			return $payload;
99			} catch (Exception $e) {
100			throw $e;
101			}
102			}
103
104
105			/**
106			* Loop de consumo de mensagem
107			*
108			* @param int $sleepSeconds
109			* @throws Exception
110			*/
111			public function consume(
112			string $queueName,
113			int $sleepSeconds = 3
114			) {
115			$this->validateExecuteCallback();
116
117
118			$this->loopConnection(function () use ($sleepSeconds, $queueName) {
119
120			$this->createQueue($queueName);
121
122			$this->channel->basic_qos(null, 1, null);
123
124			$this->channel->basic_consume(
125			$this->queueName,
126			$this->randomConsumer(),
127			false,
128			false,
129			false,
130			false,
131			function (AMQPMessage $message) {
132			pcntl_sigprocmask(SIG_BLOCK, [SIGTERM, SIGINT]);
133
134			$incomeData = json_decode($message->getBody(), true);
135
136			//se o status for negativo, não executa o consumo
137			$statusBoolean = $this->executeStatusCallback($message);
138
139			if (!$statusBoolean) {
140			return false;
141			}
142
143			$receiveBoolean = $this->executeReceiveCallback($message, $incomeData);
144
145			if (!$receiveBoolean) {
146			return false;
147			}
148
149			try {
150			$this->executeMessage($message, $incomeData);
151			} catch (Exception $e) {
152			$message->nack(true);
153			$this->executeErrorCallback($e, $incomeData);
154			}
155
156			pcntl_sigprocmask(SIG_UNBLOCK, [SIGTERM, SIGINT]);
157			}
158			);
159
160			// Loop as long as the channel has callbacks registered
161			while ($this->channel->is_open()) {
162			$this->channel->wait(null, false);
163			sleep($sleepSeconds);
164
165			// Despachar eventos de "finalizar" script
166			pcntl_signal_dispatch();
167			}
168			});
169			}
170
171			public function onCheckStatus(\Closure $callback)
172			{
173			$this->onCheckStatusCallback = $callback;
174			}
175
176			public function onReceive(\Closure $callback)
177			{
178			$this->onReceiveCallback = $callback;
179			}
180
181			public function onExecuting(\Closure $callback)
182			{
183			$this->onExecutingCallback = $callback;
184			}
185
186			public function onError(\Closure $callback)
187			{
188			$this->onErrorCallback = $callback;
189			}
190
191			public function executeStatusCallback(AMQPMessage $message)
192			{
193			if (empty($this->onCheckStatusCallback)) {
194			return true;
195			}
196
197			$checkStatusCallback = $this->onCheckStatusCallback;
198			$statusBoolean = $checkStatusCallback();
199
200			if (!$statusBoolean && isset($statusBoolean)) {
201			print_r("[WORKER STOPPED]" . PHP_EOL);
202			$message->nack(true);
203			pcntl_sigprocmask(SIG_UNBLOCK, [SIGTERM, SIGINT]);
204			return false;
205			}
206
207			return true;
208			}
209
210			public function executeReceiveCallback(AMQPMessage $message, $incomeData)
211			{
212			if (empty($this->onReceiveCallback)) {
213			return true;
214			}
215
216			$receiveCallback = $this->onReceiveCallback;
217			$statusBoolean = $receiveCallback($incomeData);
218			if (!$statusBoolean && isset($statusBoolean)) {
219			print_r("[TASK IGNORED BY ON RECEIVE RETURN]" . PHP_EOL);
220			$message->nack();
221			pcntl_sigprocmask(SIG_UNBLOCK, [SIGTERM, SIGINT]);
222			return false;
223			}
224
225
226			return true;
227			}
228
229			public function executeMessage(AMQPMessage $message, $incomeData)
230			{
231			$onExecutingCallback = $this->onExecutingCallback;
232			$onExecutingCallback($message, $incomeData);
233			}
234
235
236			public function executeErrorCallback(\Exception $e, $incomeData)
237			{
238			if (empty($this->onErrorCallback)) {
239			return false;
240			}
241
242			$errorCallback = $this->onErrorCallback;
243			$errorCallback($e, $incomeData);
244			return true;
245			}
246
247			public function validateExecuteCallback()
248			{
249			if (!empty($this->onExecutingCallback)) {
250			return true;
251			}
252
253			throw new \Exception("Define a onExecuting callback");
254			}
255			}
256

WillRy / rabbitrun

Push — main ( 1f8a6a...d3094e )

Queue::consume() A

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like