Consumer::take() - Code Metrics - Inspection of "add service container" - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( f55c18...052a19 )

by Baptiste

created 2018-02-12 06:55 UTC

Consumer::take() A

↳ Parent: Consumer

Complexity

Conditions	1
Paths	1

Size

Total Lines	5
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nc	1
nop	1
dl	0
loc	5
ccs	3
cts	3
cp	1
crap	1
rs	9.4285
c	0
b	0
f	0

<?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,
filter:
    dependency_paths: ["lib/*"]
    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);

                $this->requeue($frame->values()->get(1)->original()->value());
            }
        } while (!$frame->is($expected));

        $this->canceled = true;
    }
}


1		<?php
		0 ignored issues – show Coding Style Compatibility introduced 2017-10-08 13:29 UTC by Report Bug Copy Issue Report For compatibility and reusability of your code, PSR1 recommends that a file should introduce either new symbols (like classes, functions, etc.) or have side-effects (like outputting something, or including other files), but not both at the same time. The first symbol is defined on line 34 and the first side effect is on line 23. The PSR-1: Basic Coding Standard recommends that a file should either introduce new symbols, that is classes, functions, constants or similar, or have side effects. Side effects are anything that executes logic, like for example printing output, changing ini settings or writing to a file. The idea behind this recommendation is that merely auto-loading a class should not change the state of an application. It also promotes a cleaner style of programming and makes your code less prone to errors, because the logic is not spread out all over the place. To learn more about the PSR-1, please see the PHP-FIG site on the PSR-1. Loading history...
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,
		0 ignored issues – show Bug introduced 2018-02-12 06:58 UTC by Report Bug Copy Issue Report The type `Innmind\AMQP\Exception\M...AnotherConsumerReceived` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
18		Exception\Requeue,
19		Exception\Reject,
20		Exception\Cancel
21		};
22
23		final class Consumer implements ConsumerInterface
		0 ignored issues – show Bug introduced 2017-10-08 13:29 UTC by Report Bug Copy Issue Report Possible parse error: class missing opening or closing brace Loading history...
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);
		0 ignored issues – show Comprehensibility Best Practice introduced 2018-02-12 06:58 UTC by Report Bug Copy Issue Report The variable `$deliveryTag` does not seem to be defined for all execution paths leading up to this point. Loading history...
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 2018-02-12 06:58 UTC by Report Bug Copy Issue Report The assignment to `$message` is dead and can be removed. 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 ( f55c18...052a19 )

Consumer::take() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like