Connection::wait() - Code Metrics - Inspection of "add possibility to lock a message" - Innmind/AMQP - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — develop ( 13fd92...a71c17 )

by Baptiste

created 2017-09-10 10:00 UTC

Connection::wait() C

↳ Parent: Connection

Complexity

Conditions	7
Paths	10

Size

Total Lines	32
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	7.2694

Importance

Changes

Metric	Value
dl	0
loc	32
ccs	14
cts	17
cp	0.8235
rs	6.7272
c	0
b	0
f	0
cc	7
eloc	18
nc	10
nop	1
crap	7.2694

<?php
declare(strict_types = 1);

namespace Innmind\AMQP\Transport;

use Innmind\AMQP\{
    Transport\Connection\FrameReader,
    Transport\Protocol\Version,
    Transport\Frame\Type,
    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();

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

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

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

Innmind / AMQP

GitHub Access Token became invalid

Push — develop ( 13fd92...a71c17 )

Connection::wait() C

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like