Connection::__destruct() - Code Metrics - Inspection of "Merge branch 'develop'" - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 3e12fc...78d3de )

by Baptiste

created 2017-11-27 07:02 UTC

Connection::__destruct() A

↳ Parent: Connection::buildSocket()

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

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

<?php
declare(strict_types = 1);

namespace Innmind\AMQP\Transport\Connection;

use Innmind\AMQP\{
    Transport\Connection as ConnectionInterface,
    Transport\Frame,
    Transport\Protocol,
    Transport\Protocol\Version,
    Transport\Frame\Type,
    Transport\Frame\Method,
    Transport\Frame\Value\UnsignedOctet,
    Model\Connection\StartOk,
    Model\Connection\SecureOk,
    Model\Connection\TuneOk,
    Model\Connection\Open,
    Model\Connection\Close,
    Model\Connection\MaxChannels,
    Model\Connection\MaxFrameSize,
    Exception\FrameChannelExceedAllowedChannelNumber,
    Exception\FrameExceedAllowedSize,
    Exception\UnexpectedFrame,
    Exception\NoFrameDetected,
    Exception\ConnectionClosed
};
use Innmind\Socket\{
    Internet\Transport,
    Client\Internet as Socket
};
use Innmind\Stream\Select;
use Innmind\Url\{
    UrlInterface,
    Authority\NullUserInformation
};
use Innmind\TimeContinuum\{
    ElapsedPeriod,
    TimeContinuumInterface
};
use Innmind\Immutable\Str;

final class Connection implements ConnectionInterface
{
    private $transport;
    private $authority;
    private $vhost;
    private $protocol;
    private $socket;
    private $timeout;
    private $select;
    private $read;
    private $closed = true;
    private $opening = true;
    private $maxChannels;
    private $maxFrameSize;
    private $heartbeat;
    private $clock;
    private $lastReceivedData;

    public function __construct(
        Transport $transport,
        UrlInterface $server,
        Protocol $protocol,
        ElapsedPeriod $timeout,
        TimeContinuumInterface $clock
    ) {
        $this->transport = $transport;
        $this->authority = $server->authority();
        $this->vhost = $server->path();
        $this->protocol = $protocol;
        $this->timeout = $timeout;
        $this->buildSocket();
        $this->read = new FrameReader;
        $this->maxChannels = new MaxChannels(0);
        $this->maxFrameSize = new MaxFrameSize(0);
        $this->heartbeat = $timeout;
        $this->clock = $clock;
        $this->lastReceivedData = $clock->now();

        $this->open();
    }

    public function protocol(): Protocol
    {
        return $this->protocol;
    }

    public function send(Frame $frame): ConnectionInterface
    {
        if (!$this->maxChannels->allows($frame->channel()->toInt())) {
            throw new FrameChannelExceedAllowedChannelNumber(
                $frame->channel(),
                $this->maxChannels
            );
        }

        $frame = (new Str((string) $frame))->toEncoding('ASCII');

        if (!$this->maxFrameSize->allows($frame->length())) {
            throw new FrameExceedAllowedSize(
                $frame->length(),
                $this->maxFrameSize
            );
        }

        $this->socket->write($frame);

        return $this;
    }

    /**
     * {@inheritdoc}
     */
    public function wait(string ...$names): Frame
    {
        do {
            if (!$this->opening && $this->closed()) {
                throw new ConnectionClosed;
            }

            $now = $this->clock->now();
            $elapsedPeriod = $now->elapsedSince($this->lastReceivedData);
function acceptsInteger($int) { }

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

// Instead of
acceptsInteger($x);

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

            if ($elapsedPeriod->longerThan($this->heartbeat)) {
                $this->send(Frame::heartbeat());
            }

            $streams = ($this->select)();
        } while (!$streams->get('read')->contains($this->socket));

        $frame = ($this->read)($this->socket, $this->protocol);
        $this->lastReceivedData = $this->clock->now();

        if ($frame->type() === Type::heartbeat()) {
            return $this->wait(...$names);
        }

        if (count($names) === 0) {
            return $frame;
        }

        if ($frame->type() !== Type::method()) {
            //someone must have forgot a wait() call
            throw new ExpectedMethodFrame($frame->type());
        }

        foreach ($names as $name) {
            if ($frame->is($this->protocol->method($name))) {
                return $frame;
            }
        }

        if ($frame->is($this->protocol->method('connection.close'))) {
            $this->send($this->protocol->connection()->closeOk());
            $this->closed = true;

            throw ConnectionClosed::byServer(
                (string) $frame->values()->get(1)->original(),
                $frame->values()->get(0)->original()->value(),
                new Method(
                    $frame->values()->get(2)->original()->value(),
                    $frame->values()->get(3)->original()->value()
                )
            );
        }

        throw new UnexpectedFrame($frame, ...$names);
    }

    public function maxFrameSize(): MaxFrameSize
    {
        return $this->maxFrameSize;
    }

    public function close(): void
    {
        if ($this->closed()) {
            return;
        }

        $this
            ->send($this->protocol->connection()->close(new Close))
            ->wait('connection.close-ok');
        $this->socket->close();
        $this->closed = true;
    }

    public function closed(): bool
    {
        return $this->closed || $this->socket->closed();
    }

    private function buildSocket(): void
    {
        $this->socket = new Socket(
            $this->transport,
            $this->authority->withUserInformation(new NullUserInformation)
        );
        $this->select = (new Select($this->timeout))->forRead($this->socket);
    }

    private function open(): void
    {
        if (!$this->closed()) {
            return;
        }

        $this->start();
        $this->handshake();
        $this->openVHost();

        $this->closed = false;
        $this->opening = false;
    }

    private function start(): void
    {
        $this->socket->write(
            new Str((string) $this->protocol->version())
        );

        try {
            $frame = $this->wait('connection.start');
$myVar = 'Value';
$higher = false;

if (rand(1, 6) > 3) {
    $higher = true;
} else {
    $higher = false;
}
        } catch (NoFrameDetected $e) {
            $content = $e->content()->toEncoding('ASCII');

            if (
                $content->length() !== 8 ||
                !$content->matches('/^AMQP/')
            ) {
                throw $e;
            }

            $version = $content
                ->substring(5, 8)
                ->chunk();
            $this->protocol->use(
                new Version(
                    UnsignedOctet::fromString($version->get(0))->original()->value(),
                    UnsignedOctet::fromString($version->get(1))->original()->value(),
                    UnsignedOctet::fromString($version->get(2))->original()->value()
                )
            );
            //socket rebuilt as the server close the connection on version mismatch
            $this->buildSocket();
            $this->start();

            return;
        }

        $this->send($this->protocol->connection()->startOk(
            new StartOk(
                $this->authority->userInformation()->user(),
                $this->authority->userInformation()->password()
            )
        ));
    }

    private function handshake(): void
    {
        $frame = $this->wait('connection.secure', 'connection.tune');

        if ($frame->is($this->protocol->method('connection.secure'))) {
            $this->send($this->protocol->connection()->secureOk(
                new SecureOk(
                    $this->authority->userInformation()->user(),
                    $this->authority->userInformation()->password()
                )
            ));
            $frame = $this->wait('connection.tune');
        }

        $this->maxChannels = new MaxChannels(
            $frame->values()->get(0)->original()->value()
        );
        $this->maxFrameSize = new MaxFrameSize(
            $frame->values()->get(1)->original()->value()
        );
        $this->heartbeat = new ElapsedPeriod(
            $frame->values()->get(2)->original()->value()
        );
        $this->select = (new Select($this->heartbeat))->forRead($this->socket);
        $this->send($this->protocol->connection()->tuneOk(
            new TuneOk(
                $this->maxChannels,
                $this->maxFrameSize,
                $this->heartbeat
            )
        ));
    }

    private function openVHost(): void
    {
        $this
            ->send($this->protocol->connection()->open(
                new Open($this->vhost)
            ))
            ->wait('connection.open-ok');
    }
}


1		<?php
2		declare(strict_types = 1);
3
4		namespace Innmind\AMQP\Transport\Connection;
5
6		use Innmind\AMQP\{
7		Transport\Connection as ConnectionInterface,
8		Transport\Frame,
9		Transport\Protocol,
10		Transport\Protocol\Version,
11		Transport\Frame\Type,
12		Transport\Frame\Method,
13		Transport\Frame\Value\UnsignedOctet,
14		Model\Connection\StartOk,
15		Model\Connection\SecureOk,
16		Model\Connection\TuneOk,
17		Model\Connection\Open,
18		Model\Connection\Close,
19		Model\Connection\MaxChannels,
20		Model\Connection\MaxFrameSize,
21		Exception\FrameChannelExceedAllowedChannelNumber,
22		Exception\FrameExceedAllowedSize,
23		Exception\UnexpectedFrame,
24		Exception\NoFrameDetected,
25		Exception\ConnectionClosed
26		};
27		use Innmind\Socket\{
28		Internet\Transport,
29		Client\Internet as Socket
30		};
31		use Innmind\Stream\Select;
32		use Innmind\Url\{
33		UrlInterface,
34		Authority\NullUserInformation
35		};
36		use Innmind\TimeContinuum\{
37		ElapsedPeriod,
38		TimeContinuumInterface
39		};
40		use Innmind\Immutable\Str;
41
42		final class Connection implements ConnectionInterface
43		{
44		private $transport;
45		private $authority;
46		private $vhost;
47		private $protocol;
48		private $socket;
49		private $timeout;
50		private $select;
51		private $read;
52		private $closed = true;
53		private $opening = true;
54		private $maxChannels;
55		private $maxFrameSize;
56		private $heartbeat;
57		private $clock;
58		private $lastReceivedData;
59
60	88	public function __construct(
61		Transport $transport,
62		UrlInterface $server,
63		Protocol $protocol,
64		ElapsedPeriod $timeout,
65		TimeContinuumInterface $clock
66		) {
67	88	$this->transport = $transport;
68	88	$this->authority = $server->authority();
69	88	$this->vhost = $server->path();
70	88	$this->protocol = $protocol;
71	88	$this->timeout = $timeout;
72	88	$this->buildSocket();
73	88	$this->read = new FrameReader;
74	88	$this->maxChannels = new MaxChannels(0);
75	88	$this->maxFrameSize = new MaxFrameSize(0);
76	88	$this->heartbeat = $timeout;
77	88	$this->clock = $clock;
78	88	$this->lastReceivedData = $clock->now();
79
80	88	$this->open();
81	88	}
82
83	80	public function protocol(): Protocol
84		{
85	80	return $this->protocol;
86		}
87
88	88	public function send(Frame $frame): ConnectionInterface
89		{
90	88	if (!$this->maxChannels->allows($frame->channel()->toInt())) {
91		throw new FrameChannelExceedAllowedChannelNumber(
92		$frame->channel(),
93		$this->maxChannels
94		);
95		}
96
97	88	$frame = (new Str((string) $frame))->toEncoding('ASCII');
98
99	88	if (!$this->maxFrameSize->allows($frame->length())) {
100		throw new FrameExceedAllowedSize(
101		$frame->length(),
102		$this->maxFrameSize
103		);
104		}
105
106	88	$this->socket->write($frame);
107
108	88	return $this;
109		}
110
111		/**
112		* {@inheritdoc}
113		*/
114	88	public function wait(string ...$names): Frame
115		{
116		do {
117	88	if (!$this->opening && $this->closed()) {
118		throw new ConnectionClosed;
119		}
120
121	88	$now = $this->clock->now();
122	88	$elapsedPeriod = $now->elapsedSince($this->lastReceivedData);
		0 ignored issues – show Documentation introduced 2017-09-09 16:02 UTC by Report Bug Copy Issue Report `$this->lastReceivedData` is of type `object<Innmind\TimeConti...m\PointInTimeInterface>`, but the function expects a `object<self>`. 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...
123
124	88	if ($elapsedPeriod->longerThan($this->heartbeat)) {
125	17	$this->send(Frame::heartbeat());
126		}
127
128	88	$streams = ($this->select)();
129	88	} while (!$streams->get('read')->contains($this->socket));
130
131	88	$frame = ($this->read)($this->socket, $this->protocol);
132	88	$this->lastReceivedData = $this->clock->now();
133
134	88	if ($frame->type() === Type::heartbeat()) {
135		return $this->wait(...$names);
136		}
137
138	88	if (count($names) === 0) {
139	32	return $frame;
140		}
141
142	88	if ($frame->type() !== Type::method()) {
143		//someone must have forgot a wait() call
144		throw new ExpectedMethodFrame($frame->type());
145		}
146
147	88	foreach ($names as $name) {
148	88	if ($frame->is($this->protocol->method($name))) {
149	88	return $frame;
150		}
151		}
152
153	4	if ($frame->is($this->protocol->method('connection.close'))) {
154	2	$this->send($this->protocol->connection()->closeOk());
155	2	$this->closed = true;
156
157	2	throw ConnectionClosed::byServer(
158	2	(string) $frame->values()->get(1)->original(),
159	2	$frame->values()->get(0)->original()->value(),
160	2	new Method(
161	2	$frame->values()->get(2)->original()->value(),
162	2	$frame->values()->get(3)->original()->value()
163		)
164		);
165		}
166
167	2	throw new UnexpectedFrame($frame, ...$names);
168		}
169
170	36	public function maxFrameSize(): MaxFrameSize
171		{
172	36	return $this->maxFrameSize;
173		}
174
175	72	public function close(): void
176		{
177	72	if ($this->closed()) {
178		return;
179		}
180
181		$this
182	72	->send($this->protocol->connection()->close(new Close))
183	72	->wait('connection.close-ok');
184	72	$this->socket->close();
185	72	$this->closed = true;
186	72	}
187
188	88	public function closed(): bool
189		{
190	88	return $this->closed \|\| $this->socket->closed();
191		}
192
193	88	private function buildSocket(): void
194		{
195	88	$this->socket = new Socket(
196	88	$this->transport,
197	88	$this->authority->withUserInformation(new NullUserInformation)
198		);
199	88	$this->select = (new Select($this->timeout))->forRead($this->socket);
200	88	}
201
202	88	private function open(): void
203		{
204	88	if (!$this->closed()) {
205		return;
206		}
207
208	88	$this->start();
209	88	$this->handshake();
210	88	$this->openVHost();
211
212	88	$this->closed = false;
213	88	$this->opening = false;
214	88	}
215
216	88	private function start(): void
217		{
218	88	$this->socket->write(
219	88	new Str((string) $this->protocol->version())
220		);
221
222		try {
223	88	$frame = $this->wait('connection.start');
		0 ignored issues – show Unused Code introduced 2017-09-03 14:48 UTC by Report Bug Copy Issue Report `$frame` 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...
224	2	} catch (NoFrameDetected $e) {
225	2	$content = $e->content()->toEncoding('ASCII');
226
227		if (
228	2	$content->length() !== 8 \|\|
229	2	!$content->matches('/^AMQP/')
230		) {
231		throw $e;
232		}
233
234		$version = $content
235	2	->substring(5, 8)
236	2	->chunk();
237	2	$this->protocol->use(
238	2	new Version(
239	2	UnsignedOctet::fromString($version->get(0))->original()->value(),
240	2	UnsignedOctet::fromString($version->get(1))->original()->value(),
241	2	UnsignedOctet::fromString($version->get(2))->original()->value()
242		)
243		);
244		//socket rebuilt as the server close the connection on version mismatch
245	2	$this->buildSocket();
246	2	$this->start();
247
248	2	return;
249		}
250
251	88	$this->send($this->protocol->connection()->startOk(
252	88	new StartOk(
253	88	$this->authority->userInformation()->user(),
254	88	$this->authority->userInformation()->password()
255		)
256		));
257	88	}
258
259	88	private function handshake(): void
260		{
261	88	$frame = $this->wait('connection.secure', 'connection.tune');
262
263	88	if ($frame->is($this->protocol->method('connection.secure'))) {
264		$this->send($this->protocol->connection()->secureOk(
265		new SecureOk(
266		$this->authority->userInformation()->user(),
267		$this->authority->userInformation()->password()
268		)
269		));
270		$frame = $this->wait('connection.tune');
271		}
272
273	88	$this->maxChannels = new MaxChannels(
274	88	$frame->values()->get(0)->original()->value()
275		);
276	88	$this->maxFrameSize = new MaxFrameSize(
277	88	$frame->values()->get(1)->original()->value()
278		);
279	88	$this->heartbeat = new ElapsedPeriod(
280	88	$frame->values()->get(2)->original()->value()
281		);
282	88	$this->select = (new Select($this->heartbeat))->forRead($this->socket);
283	88	$this->send($this->protocol->connection()->tuneOk(
284	88	new TuneOk(
285	88	$this->maxChannels,
286	88	$this->maxFrameSize,
287	88	$this->heartbeat
288		)
289		));
290	88	}
291
292	88	private function openVHost(): void
293		{
294		$this
295	88	->send($this->protocol->connection()->open(
296	88	new Open($this->vhost)
297		))
298	88	->wait('connection.open-ok');
299	88	}
300		}
301

Innmind / AMQP

GitHub Access Token became invalid

Push — master ( 3e12fc...78d3de )

Connection::__destruct() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like