Connection::opened() - Code Metrics - Inspection of "send heartbeat frame when no data received for a c..." - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( d40077...e71353 )

by Baptiste

created 2017-09-09 15:59 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\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,
    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();

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

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

Innmind / AMQP

GitHub Access Token became invalid

Push — develop ( d40077...e71353 )

Connection::opened() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like