Connection::opened() - Code Metrics - Inspection of "add basic client" - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( a71c17...924269 )

by Baptiste

created 2017-09-10 16:56 UTC

Connection::opened() A

↳ Parent: Connection

Complexity

Conditions	2
Paths	2

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	2

Importance

Changes

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

<?php
declare(strict_types = 1);

namespace Innmind\AMQP\Transport;

use Innmind\AMQP\{
    Transport\Connection\FrameReader,
    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
{
    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;
    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): 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 {
            $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 ($this->protocol->method($name)->equals($frame->method())) {
                return $frame;
            }
        }

        if ($this->protocol->method('connection.close')->equals($frame->method())) {
            $this->send($this->protocol->connection()->closeOk());
            $this->opened = false;

            throw new ConnectionClosed(
                (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->method(), ...$names);
    }

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

    public 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;
    }

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

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

Innmind / AMQP

GitHub Access Token became invalid

Push — develop ( a71c17...924269 )

Connection::opened() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like