Consumer::__construct() - Code Metrics - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Consumer::__construct() A
last analyzed 2019-02-26 20:40 UTC

↳ Parent: Consumer

Complexity

Conditions	1
Paths	1

Size

Total Lines	13
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	7
nc	1
nop	4
dl	0
loc	13
ccs	7
cts	7
cp	1
crap	1
rs	10
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
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
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	14	public function __construct(
35		Connection $connection,
36		Consume $command,
37		Channel $channel,
38		string $consumerTag
39		) {
40	14	$this->connection = $connection;
41	14	$this->command = $command;
42	14	$this->channel = $channel;
43	14	$this->consumerTag = $consumerTag;
44	14	$this->read = new MessageReader;
45		$this->predicate = static function(): bool {
46	12	return true; //by default consume all messages
47		};
48	14	}
49
50		/**
51		* {@inheritdoc}
52		*/
53	14	public function foreach(callable $consume): void
54		{
55	14	if ($this->canceled) {
56	4	return;
57		}
58
59	14	while ($this->shouldConsume()) {
60		try {
61	14	$frame = $this->connection->wait('basic.deliver');
62	14	$message = ($this->read)($this->connection);
63	14	$message = new Locked($message);
64	14	$consumerTag = (string) $frame->values()->first()->original();
65	14	$deliveryTag = $frame->values()->get(1)->original()->value();
66	14	$redelivered = $frame->values()->get(2)->original()->first();
67	14	$exchange = (string) $frame->values()->get(3)->original();
68	14	$routingKey = (string) $frame->values()->get(4)->original();
69
70	14	if ($this->consumerTag !== $consumerTag) {
71		throw new MessageFromAnotherConsumerReceived(
72		$message,
73		$consumerTag,
74		$deliveryTag,
75		$redelivered,
76		$exchange,
77		$routingKey
78		);
79		}
80
81	14	$toProcess = ($this->predicate)(
82	14	$message,
83	7	$redelivered,
84	7	$exchange,
85	7	$routingKey
86		);
87
88	14	if (!$toProcess) {
89	2	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	14	$this->flagConsumed();
96
97	14	$consume(
98	14	$message,
99	7	$redelivered,
100	7	$exchange,
101	7	$routingKey
102		);
103
104	10	$this->ack($deliveryTag);
105	10	} catch (Reject $e) {
106	2	$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	8	} catch (Requeue $e) {
108	4	$this->requeue($deliveryTag);
109	4	} catch (Cancel $e) {
110	2	$this->ack($deliveryTag);
111
112	2	break;
113	2	} catch (\Throwable $e) {
114	2	$this->requeue($deliveryTag);
115	2	$this->cancel();
116
117	2	throw $e;
118		}
119		}
120
121	12	$this->cancel();
122	12	}
123
124		/**
125		* {@inheritdoc}
126		*/
127	10	public function take(int $count): ConsumerInterface
128		{
129	10	$this->take = $count;
130
131	10	return $this;
132		}
133
134		/**
135		* {@inheritdoc}
136		*/
137	2	public function filter(callable $predicate): ConsumerInterface
138		{
139	2	$this->predicate = $predicate;
140
141	2	return $this;
142		}
143
144	14	private function shouldConsume(): bool
145		{
146	14	if ($this->canceled()) {
147		return false;
148		}
149
150	14	if (\is_null($this->take)) {
151	4	return true;
152		}
153
154	10	return $this->take > 0;
155		}
156
157	14	private function flagConsumed(): void
158		{
159	14	if (\is_null($this->take)) {
160	4	return;
161		}
162
163	10	--$this->take;
164	10	}
165
166	10	private function ack(int $deliveryTag): void
167		{
168	10	if ($this->command->shouldAutoAcknowledge()) {
169		return;
170		}
171
172	10	$this->connection->send(
173	10	$this->connection->protocol()->basic()->ack(
174	10	$this->channel,
175	10	new Ack($deliveryTag)
176		)
177		);
178	10	}
179
180	2	private function reject(int $deliveryTag): void
181		{
182	2	$this->connection->send(
183	2	$this->connection->protocol()->basic()->reject(
184	2	$this->channel,
185	2	new RejectCommand($deliveryTag)
186		)
187		);
188	2	}
189
190	10	private function requeue(int $deliveryTag): void
191		{
192	10	$this->connection->send(
193	10	$this->connection->protocol()->basic()->reject(
194	10	$this->channel,
195	10	RejectCommand::requeue($deliveryTag)
196		)
197		);
198	10	}
199
200	14	private function canceled(): bool
201		{
202	14	if ($this->canceled) {
203		return true;
204		}
205
206	14	return $this->connection->closed();
207		}
208
209	14	private function cancel(): void
210		{
211	14	if ($this->canceled()) {
212		return;
213		}
214
215	14	$this->connection->send($this->connection->protocol()->basic()->cancel(
216	14	$this->channel,
217	14	new CancelCommand($this->consumerTag)
218		));
219
220	14	$deliver = $this->connection->protocol()->method('basic.deliver');
221	14	$expected = $this->connection->protocol()->method('basic.cancel-ok');
222
223		do {
224	14	$frame = $this->connection->wait();
225
226		if (
227	14	$frame->type() === Type::method() &&
228	14	$frame->is($deliver)
229		) {
230		//requeue all the messages sent right before the cancel method
231	10	$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	10	$this->requeue($frame->values()->get(1)->original()->value());
233		}
234	14	} while (!$frame->is($expected));
235
236	14	$this->canceled = true;
237	14	}
238		}
239

Innmind / AMQP

GitHub Access Token became invalid

Consumer::__construct() A last analyzed 2019-02-26 20:40 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Consumer::__construct() A
last analyzed 2019-02-26 20:40 UTC