Connection::wait() - Code Metrics - Inspection of "fix protocol selection when multiple ones are supp..." - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( c3bf76...dc595a )

by Baptiste

created 2017-09-03 14:45 UTC

Connection::wait() A

↳ Parent: Connection

Complexity

Conditions	4
Paths	3

Size

Total Lines	16
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	4

Importance

Changes

Metric	Value
dl	0
loc	16
ccs	8
cts	8
cp	1
rs	9.2
c	0
b	0
f	0
cc	4
eloc	9
nc	3
nop	1
crap	4

<?php
declare(strict_types = 1);

namespace Innmind\AMQP\Transport;

use Innmind\AMQP\{
    Transport\Connection\FrameReader,
    Transport\Protocol\Version,
    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
};
use Innmind\Socket\{
    Internet\Transport,
    Client\Internet as Socket
};
use Innmind\Stream\Select;
use Innmind\Url\{
    UrlInterface,
    Authority\NullUserInformation
};
use Innmind\TimeContinuum\ElapsedPeriod;
use Innmind\Immutable\Str;

final class Connection
{
    private $transport;
    private $authority;
    private $vhost;
    private $protocol;
    private $socket;
    private $timeout;
    private $select;
    private $read;
    private $opened = false;
    private $maxChannels;
    private $maxFrameSize;
    private $heartbeat;

    public function __construct(
        Transport $transport,
        UrlInterface $server,
        Protocol $protocol,
        ElapsedPeriod $timeout
    ) {
        $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->open();
    }

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

    public function send(Frame $frame): self
    {
        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;
    }

    public function wait(string ...$names): Frame
    {
        do {
            $streams = ($this->select)();
        } while (!$streams->get('read')->contains($this->socket));

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

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

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

    public function __destruct()
    {
        $this->close();
    }

    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->opened()) {
            return;
        }

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

        $this->opened = true;
    }

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

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

    private function opened(): bool
    {
        return $this->opened && !$this->socket->closed();
    }

    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 ($this->protocol->method('connection.secure')->equals($frame->method())) {
            $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;
5
6		use Innmind\AMQP\{
7		Transport\Connection\FrameReader,
8		Transport\Protocol\Version,
9		Transport\Frame\Value\UnsignedOctet,
10		Model\Connection\StartOk,
11		Model\Connection\SecureOk,
12		Model\Connection\TuneOk,
13		Model\Connection\Open,
14		Model\Connection\Close,
15		Model\Connection\MaxChannels,
16		Model\Connection\MaxFrameSize,
17		Exception\FrameChannelExceedAllowedChannelNumber,
18		Exception\FrameExceedAllowedSize,
19		Exception\UnexpectedFrame,
20		Exception\NoFrameDetected
21		};
22		use Innmind\Socket\{
23		Internet\Transport,
24		Client\Internet as Socket
25		};
26		use Innmind\Stream\Select;
27		use Innmind\Url\{
28		UrlInterface,
29		Authority\NullUserInformation
30		};
31		use Innmind\TimeContinuum\ElapsedPeriod;
32		use Innmind\Immutable\Str;
33
34		final class Connection
35		{
36		private $transport;
37		private $authority;
38		private $vhost;
39		private $protocol;
40		private $socket;
41		private $timeout;
42		private $select;
43		private $read;
44		private $opened = false;
45		private $maxChannels;
46		private $maxFrameSize;
47		private $heartbeat;
48
49	3	public function __construct(
50		Transport $transport,
51		UrlInterface $server,
52		Protocol $protocol,
53		ElapsedPeriod $timeout
54		) {
55	3	$this->transport = $transport;
56	3	$this->authority = $server->authority();
57	3	$this->vhost = $server->path();
58	3	$this->protocol = $protocol;
59	3	$this->timeout = $timeout;
60	3	$this->buildSocket();
61	3	$this->read = new FrameReader;
62	3	$this->maxChannels = new MaxChannels(0);
63	3	$this->maxFrameSize = new MaxFrameSize(0);
64	3	$this->heartbeat = $timeout;
65
66	3	$this->open();
67	3	}
68
69	2	public function protocol(): Protocol
70		{
71	2	return $this->protocol;
72		}
73
74	3	public function send(Frame $frame): self
75		{
76	3	if (!$this->maxChannels->allows($frame->channel()->toInt())) {
77		throw new FrameChannelExceedAllowedChannelNumber(
78		$frame->channel(),
79		$this->maxChannels
80		);
81		}
82
83	3	$frame = (new Str((string) $frame))->toEncoding('ASCII');
84
85	3	if (!$this->maxFrameSize->allows($frame->length())) {
86		throw new FrameExceedAllowedSize(
87		$frame->length(),
88		$this->maxFrameSize
89		);
90		}
91
92	3	$this->socket->write($frame);
93
94	3	return $this;
95		}
96
97	3	public function wait(string ...$names): Frame
98		{
99		do {
100	3	$streams = ($this->select)();
101	3	} while (!$streams->get('read')->contains($this->socket));
102
103	3	$frame = ($this->read)($this->socket, $this->protocol);
104
105	3	foreach ($names as $name) {
106	3	if ($this->protocol->method($name)->equals($frame->method())) {
107	3	return $frame;
108		}
109		}
110
111	1	throw new UnexpectedFrame($frame->method(), ...$names);
112		}
113
114	3	public function __destruct()
115		{
116	3	$this->close();
117	3	}
118
119	3	private function buildSocket(): void
120		{
121	3	$this->socket = new Socket(
122	3	$this->transport,
123	3	$this->authority->withUserInformation(new NullUserInformation)
124		);
125	3	$this->select = (new Select($this->timeout))->forRead($this->socket);
126	3	}
127
128	3	private function open(): void
129		{
130	3	if ($this->opened()) {
131		return;
132		}
133
134	3	$this->start();
135	3	$this->handshake();
136	3	$this->openVHost();
137
138	3	$this->opened = true;
139	3	}
140
141	3	private function close(): void
142		{
143	3	if (!$this->opened()) {
144		return;
145		}
146
147		$this
148	3	->send($this->protocol->connection()->close(new Close))
149	3	->wait('connection.close-ok');
150	3	$this->socket->close();
151	3	$this->opened = false;
152	3	}
153
154	3	private function opened(): bool
155		{
156	3	return $this->opened && !$this->socket->closed();
157		}
158
159	3	private function start(): void
160		{
161	3	$this->socket->write(
162	3	new Str((string) $this->protocol->version())
163		);
164
165		try {
166	3	$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...
167	1	} catch (NoFrameDetected $e) {
168	1	$content = $e->content()->toEncoding('ASCII');
169
170		if (
171	1	$content->length() !== 8 \|\|
172	1	!$content->matches('/^AMQP/')
173		) {
174		throw $e;
175		}
176
177		$version = $content
178	1	->substring(5, 8)
179	1	->chunk();
180	1	$this->protocol->use(
181	1	new Version(
182	1	UnsignedOctet::fromString($version->get(0))->original()->value(),
183	1	UnsignedOctet::fromString($version->get(1))->original()->value(),
184	1	UnsignedOctet::fromString($version->get(2))->original()->value()
185		)
186		);
187		//socket rebuilt as the server close the connection on version mismatch
188	1	$this->buildSocket();
189	1	$this->start();
190
191	1	return;
192		}
193
194	3	$this->send($this->protocol->connection()->startOk(
195	3	new StartOk(
196	3	$this->authority->userInformation()->user(),
197	3	$this->authority->userInformation()->password()
198		)
199		));
200	3	}
201
202	3	private function handshake(): void
203		{
204	3	$frame = $this->wait('connection.secure', 'connection.tune');
205
206	3	if ($this->protocol->method('connection.secure')->equals($frame->method())) {
207		$this->send($this->protocol->connection()->secureOk(
208		new SecureOk(
209		$this->authority->userInformation()->user(),
210		$this->authority->userInformation()->password()
211		)
212		));
213		$frame = $this->wait('connection.tune');
214		}
215
216	3	$this->maxChannels = new MaxChannels(
217	3	$frame->values()->get(0)->original()->value()
218		);
219	3	$this->maxFrameSize = new MaxFrameSize(
220	3	$frame->values()->get(1)->original()->value()
221		);
222	3	$this->heartbeat = new ElapsedPeriod(
223	3	$frame->values()->get(2)->original()->value()
224		);
225	3	$this->select = (new Select($this->heartbeat))->forRead($this->socket);
226	3	$this->send($this->protocol->connection()->tuneOk(
227	3	new TuneOk(
228	3	$this->maxChannels,
229	3	$this->maxFrameSize,
230	3	$this->heartbeat
231		)
232		));
233	3	}
234
235	3	private function openVHost(): void
236		{
237		$this
238	3	->send($this->protocol->connection()->open(
239	3	new Open($this->vhost)
240		))
241	3	->wait('connection.open-ok');
242	3	}
243		}
244

Innmind / AMQP

GitHub Access Token became invalid

Push — develop ( c3bf76...dc595a )

Connection::wait() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like