Consumer::ack() - Code Metrics - Inspection of "add logger connection" - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( 44f5b4...dc3756 )

by Baptiste

created 2017-09-12 05:53 UTC

Consumer::ack() A

↳ Parent: Consumer

Complexity

Conditions	2
Paths	2

Size

Total Lines	13
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	2.0078

Importance

Changes

Metric	Value
dl	0
loc	13
ccs	7
cts	8
cp	0.875
rs	9.4285
c	0
b	0
f	0
cc	2
eloc	7
nc	2
nop	1
crap	2.0078

<?php
declare(strict_types = 1);

namespace Innmind\AMQP\Client\Channel\Basic\Consumer;

use Innmind\AMQP\{
    Client\Channel\Basic\Consumer as ConsumerInterface,
    Model\Basic\Consume,
    Model\Basic\Ack,
    Model\Basic\Reject as RejectCommand,
    Model\Basic\Cancel as CancelCommand,
    Model\Basic\Message\Locked,
    Transport\Connection,
    Transport\Connection\MessageReader,
    Transport\Frame\Channel,
    Transport\Frame\Type,
    Exception\MessageFromAnotherConsumerReceived,
    Exception\Requeue,
    Exception\Reject,
    Exception\Cancel
};

final class Consumer implements ConsumerInterface
{
    private $connection;
    private $command;
    private $channel;
    private $consumerTag;
    private $read;
    private $take;
    private $predicate;
    private $canceled = false;

    public function __construct(
        Connection $connection,
        Consume $command,
        Channel $channel,
        string $consumerTag
    ) {
        $this->connection = $connection;
        $this->command = $command;
        $this->channel = $channel;
        $this->consumerTag = $consumerTag;
        $this->read = new MessageReader;
        $this->predicate = static function(): bool {
            return true; //by default consume all messages
        };
    }

    /**
     * {@inheritdoc}
     */
    public function foreach(callable $consume): void
    {
        if ($this->canceled) {
            return;
        }

        while ($this->shouldConsume()) {
            try {
                $frame = $this->connection->wait('basic.deliver');
                $message = ($this->read)($this->connection);
                $message = new Locked($message);
                $consumerTag = (string) $frame->values()->first()->original();
                $deliveryTag = $frame->values()->get(1)->original()->value();
                $redelivered = $frame->values()->get(2)->original()->first();
                $exchange = (string) $frame->values()->get(3)->original();
                $routingKey = (string) $frame->values()->get(4)->original();

                if ($this->consumerTag !== $consumerTag) {
                    throw new MessageFromAnotherConsumerReceived(
                        $message,
                        $consumerTag,
                        $deliveryTag,
                        $redelivered,
                        $exchange,
                        $routingKey
                    );
                }

                $toProcess = ($this->predicate)(
                    $message,
                    $redelivered,
                    $exchange,
                    $routingKey
                );

                if (!$toProcess) {
                    throw new Requeue;
                }

                //flag before the consume call as the "take" applies to the number
                //of messages that match the predicate and the number of
                //successfully consumed messages
                $this->flagConsumed();

                $consume(
                    $message,
                    $redelivered,
                    $exchange,
                    $routingKey
                );

                $this->ack($deliveryTag);
            } catch (Reject $e) {
                $this->reject($deliveryTag);
            } catch (Requeue $e) {
                $this->requeue($deliveryTag);
            } catch (Cancel $e) {
                $this->ack($deliveryTag);

                break;
            } catch (\Throwable $e) {
                $this->requeue($deliveryTag);
                $this->cancel();

                throw $e;
            }
        }

        $this->cancel();
    }

    /**
     * {@inheritdoc}
     */
    public function take(int $count): ConsumerInterface
    {
        $this->take = $count;

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function filter(callable $predicate): ConsumerInterface
    {
        $this->predicate = $predicate;

        return $this;
    }

    private function shouldConsume(): bool
    {
        if ($this->canceled()) {
            return false;
        }

        if (is_null($this->take)) {
            return true;
        }

        return $this->take > 0;
    }

    private function flagConsumed(): void
    {
        if (is_null($this->take)) {
            return;
        }

        --$this->take;
    }

    private function ack(int $deliveryTag): void
    {
        if ($this->command->shouldAutoAcknowledge()) {
            return;
        }

        $this->connection->send(
            $this->connection->protocol()->basic()->ack(
                $this->channel,
                new Ack($deliveryTag)
            )
        );
    }

    private function reject(int $deliveryTag): void
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->reject(
                $this->channel,
                new RejectCommand($deliveryTag)
            )
        );
    }

    private function requeue(int $deliveryTag): void
    {
        $this->connection->send(
            $this->connection->protocol()->basic()->reject(
                $this->channel,
                RejectCommand::requeue($deliveryTag)
            )
        );
    }

    private function canceled(): bool
    {
        if ($this->canceled) {
            return true;
        }

        return $this->connection->closed();
    }

    private function cancel(): void
    {
        if ($this->canceled()) {
            return;
        }

        $this->connection->send($this->connection->protocol()->basic()->cancel(
            $this->channel,
            new CancelCommand($this->consumerTag)
        ));

        $deliver = $this->connection->protocol()->method('basic.deliver');
        $expected = $this->connection->protocol()->method('basic.cancel-ok');

        do {
            $frame = $this->connection->wait();

            if (
                $frame->type() === Type::method() &&
                $frame->is($deliver)
            ) {
                //requeue all the messages sent right before the cancel method
                $message = ($this->read)($this->connection);
$myVar = 'Value';
$higher = false;

if (rand(1, 6) > 3) {
    $higher = true;
} else {
    $higher = false;
}
                $this->requeue($frame->values()->get(1)->original()->value());
            }
        } while (!$frame->is($expected));

        $this->canceled = true;
    }
}


1		<?php
2		declare(strict_types = 1);
3
4		namespace Innmind\AMQP\Client\Channel\Basic\Consumer;
5
6		use Innmind\AMQP\{
7		Client\Channel\Basic\Consumer as ConsumerInterface,
8		Model\Basic\Consume,
9		Model\Basic\Ack,
10		Model\Basic\Reject as RejectCommand,
11		Model\Basic\Cancel as CancelCommand,
12		Model\Basic\Message\Locked,
13		Transport\Connection,
14		Transport\Connection\MessageReader,
15		Transport\Frame\Channel,
16		Transport\Frame\Type,
17		Exception\MessageFromAnotherConsumerReceived,
18		Exception\Requeue,
19		Exception\Reject,
20		Exception\Cancel
21		};
22
23		final class Consumer implements ConsumerInterface
24		{
25		private $connection;
26		private $command;
27		private $channel;
28		private $consumerTag;
29		private $read;
30		private $take;
31		private $predicate;
32		private $canceled = false;
33
34	7	public function __construct(
35		Connection $connection,
36		Consume $command,
37		Channel $channel,
38		string $consumerTag
39		) {
40	7	$this->connection = $connection;
41	7	$this->command = $command;
42	7	$this->channel = $channel;
43	7	$this->consumerTag = $consumerTag;
44	7	$this->read = new MessageReader;
45	6	$this->predicate = static function(): bool {
46	6	return true; //by default consume all messages
47		};
48	7	}
49
50		/**
51		* {@inheritdoc}
52		*/
53	7	public function foreach(callable $consume): void
54		{
55	7	if ($this->canceled) {
56	2	return;
57		}
58
59	7	while ($this->shouldConsume()) {
60		try {
61	7	$frame = $this->connection->wait('basic.deliver');
62	7	$message = ($this->read)($this->connection);
63	7	$message = new Locked($message);
64	7	$consumerTag = (string) $frame->values()->first()->original();
65	7	$deliveryTag = $frame->values()->get(1)->original()->value();
66	7	$redelivered = $frame->values()->get(2)->original()->first();
67	7	$exchange = (string) $frame->values()->get(3)->original();
68	7	$routingKey = (string) $frame->values()->get(4)->original();
69
70	7	if ($this->consumerTag !== $consumerTag) {
71		throw new MessageFromAnotherConsumerReceived(
72		$message,
73		$consumerTag,
74		$deliveryTag,
75		$redelivered,
76		$exchange,
77		$routingKey
78		);
79		}
80
81	7	$toProcess = ($this->predicate)(
82	7	$message,
83	7	$redelivered,
84	7	$exchange,
85	7	$routingKey
86		);
87
88	7	if (!$toProcess) {
89	1	throw new Requeue;
90		}
91
92		//flag before the consume call as the "take" applies to the number
93		//of messages that match the predicate and the number of
94		//successfully consumed messages
95	7	$this->flagConsumed();
96
97	7	$consume(
98	7	$message,
99	7	$redelivered,
100	7	$exchange,
101	7	$routingKey
102		);
103
104	5	$this->ack($deliveryTag);
105	5	} catch (Reject $e) {
106	1	$this->reject($deliveryTag);
107	4	} catch (Requeue $e) {
108	2	$this->requeue($deliveryTag);
109	2	} catch (Cancel $e) {
110	1	$this->ack($deliveryTag);
111
112	1	break;
113	1	} catch (\Throwable $e) {
114	1	$this->requeue($deliveryTag);
115	1	$this->cancel();
116
117	1	throw $e;
118		}
119		}
120
121	6	$this->cancel();
122	6	}
123
124		/**
125		* {@inheritdoc}
126		*/
127	5	public function take(int $count): ConsumerInterface
128		{
129	5	$this->take = $count;
130
131	5	return $this;
132		}
133
134		/**
135		* {@inheritdoc}
136		*/
137	1	public function filter(callable $predicate): ConsumerInterface
138		{
139	1	$this->predicate = $predicate;
140
141	1	return $this;
142		}
143
144	7	private function shouldConsume(): bool
145		{
146	7	if ($this->canceled()) {
147		return false;
148		}
149
150	7	if (is_null($this->take)) {
151	2	return true;
152		}
153
154	5	return $this->take > 0;
155		}
156
157	7	private function flagConsumed(): void
158		{
159	7	if (is_null($this->take)) {
160	2	return;
161		}
162
163	5	--$this->take;
164	5	}
165
166	5	private function ack(int $deliveryTag): void
167		{
168	5	if ($this->command->shouldAutoAcknowledge()) {
169		return;
170		}
171
172	5	$this->connection->send(
173	5	$this->connection->protocol()->basic()->ack(
174	5	$this->channel,
175	5	new Ack($deliveryTag)
176		)
177		);
178	5	}
179
180	1	private function reject(int $deliveryTag): void
181		{
182	1	$this->connection->send(
183	1	$this->connection->protocol()->basic()->reject(
184	1	$this->channel,
185	1	new RejectCommand($deliveryTag)
186		)
187		);
188	1	}
189
190	5	private function requeue(int $deliveryTag): void
191		{
192	5	$this->connection->send(
193	5	$this->connection->protocol()->basic()->reject(
194	5	$this->channel,
195	5	RejectCommand::requeue($deliveryTag)
196		)
197		);
198	5	}
199
200	7	private function canceled(): bool
201		{
202	7	if ($this->canceled) {
203		return true;
204		}
205
206	7	return $this->connection->closed();
207		}
208
209	7	private function cancel(): void
210		{
211	7	if ($this->canceled()) {
212		return;
213		}
214
215	7	$this->connection->send($this->connection->protocol()->basic()->cancel(
216	7	$this->channel,
217	7	new CancelCommand($this->consumerTag)
218		));
219
220	7	$deliver = $this->connection->protocol()->method('basic.deliver');
221	7	$expected = $this->connection->protocol()->method('basic.cancel-ok');
222
223		do {
224	7	$frame = $this->connection->wait();
225
226		if (
227	7	$frame->type() === Type::method() &&
228	7	$frame->is($deliver)
229		) {
230		//requeue all the messages sent right before the cancel method
231	5	$message = ($this->read)($this->connection);
		0 ignored issues – show Unused Code introduced 2017-09-10 16:58 UTC by Report Bug Copy Issue Report `$message` is not used, you could remove the assignment. This check looks for variable assignements that are either overwritten by other assignments or where the variable is not used subsequently. $myVar = 'Value'; $higher = false; if (rand(1, 6) > 3) { $higher = true; } else { $higher = false; } Both the `$myVar` assignment in line 1 and the `$higher` assignment in line 2 are dead. The first because `$myVar` is never used and the second because `$higher` is always overwritten for every possible time line. Loading history...
232	5	$this->requeue($frame->values()->get(1)->original()->value());
233		}
234	7	} while (!$frame->is($expected));
235
236	7	$this->canceled = true;
237	7	}
238		}
239

Innmind / AMQP

GitHub Access Token became invalid

Push — develop ( 44f5b4...dc3756 )

Consumer::ack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like