Subscribe::loop() - Code Metrics - Inspection of "Assign correct blocking mode on loop" - unreal4u/mqtt - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( c8a884...c0f80b )

by Camilo

created 2018-01-03 01:33 UTC

Subscribe::loop() A

↳ Parent: Subscribe

Complexity

Conditions	3
Paths	3

Size

Total Lines	21
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
dl	0
loc	21
ccs	0
cts	8
cp	0
rs	9.3142
c	0
b	0
f	0
cc	3
eloc	12
nc	3
nop	2
crap	12

<?php

declare(strict_types=1);

namespace unreal4u\MQTT\Protocol;

use unreal4u\MQTT\Application\EmptyReadableResponse;
use unreal4u\MQTT\Internals\ClientInterface;
use unreal4u\MQTT\Internals\EventManager;
use unreal4u\MQTT\Internals\ProtocolBase;
use unreal4u\MQTT\Internals\ReadableContentInterface;
use unreal4u\MQTT\Internals\WritableContent;
use unreal4u\MQTT\Internals\WritableContentInterface;
use unreal4u\MQTT\Protocol\Subscribe\Topic;
use unreal4u\MQTT\Utilities;

final class Subscribe extends ProtocolBase implements WritableContentInterface
{
    use WritableContent;

    const CONTROL_PACKET_VALUE = 8;

    private $packetIdentifier = 0;

    /**
     * An array of topics on which to subscribe to
     * @var Topic[]
     */
    private $topics = [];

    /**
     * @return string
     * @throws \OutOfRangeException
     * @throws \Exception
     */
    public function createVariableHeader(): string
    {
        // Subscribe must always send a 2 flag
        $this->specialFlags = 2;

        // Assign a packet identifier automatically if none has been assigned yet
        if ($this->packetIdentifier === 0) {
            $this->setPacketIdentifier(random_int(0, 65535));
        }

        return Utilities::convertNumberToBinaryString($this->packetIdentifier);
    }

    public function createPayload(): string
    {
        $output = '';
        foreach ($this->topics as $topic) {
            // chr on QoS level is safe because it will create an 8-bit flag where the first 6 are only 0's
            $output .= $this->createUTF8String($topic->getTopicName()) . \chr($topic->getTopicQoSLevel());
        }
        return $output;
    }

    /**
     * When the Server receives a SUBSCRIBE Packet from a Client, the Server MUST respond with a SUBACK Packet
     *
     * This can however not be in the same order, as we may be able to receive PUBLISH packets before getting a SUBACK
     * back
     *
     * @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718134 (MQTT-3.8.4-1)
     * @return bool
     */
    public function shouldExpectAnswer(): bool
    {
        return true;
    }

    /**
     * SUBSCRIBE Control Packets MUST contain a non-zero 16-bit Packet Identifier
     *
     * @param int $packetIdentifier
     * @return Subscribe
     * @throws \OutOfRangeException
     */
    public function setPacketIdentifier(int $packetIdentifier): Subscribe
    {
        if ($packetIdentifier > 65535 || $packetIdentifier < 1) {
            throw new \OutOfRangeException('Packet identifier must fit within 2 bytes');
        }

        $this->packetIdentifier = $packetIdentifier;

        return $this;
    }

    public function getPacketIdentifier(): int
    {
        return $this->packetIdentifier;
    }

    /**
     * Performs a check on the socket connection and returns either the contents or an empty object
     *
     * @param ClientInterface $client
     * @return ReadableContentInterface
     * @throws \DomainException
     * @throws \unreal4u\MQTT\Exceptions\NotConnected
     * @throws \unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined
     */
    public function checkForEvent(ClientInterface $client): ReadableContentInterface
    {
        $this->checkPingTime($client);
        $publishPacketControlField = $client->readSocketData(1);
        $eventManager = new EventManager($this->logger);

        $this->logger->debug('Checking event', ['ordValue' => \ord($publishPacketControlField) & 255]);
        if ((\ord($publishPacketControlField) & 255) > 0) {
            return $eventManager->analyzeHeaders($publishPacketControlField, $client);
        }

        $this->logger->debug('No valid publish packet control field found, returning empty response');
        return new EmptyReadableResponse($this->logger);
    }

    /**
     * Loop and yields different type of results back whenever they are available
     *
     * @param ClientInterface $client
     * @param int $idleMicroseconds The amount of microseconds the watcher should wait before checking the socket again
     * @return \Generator
     * @throws \DomainException
     * @throws \unreal4u\MQTT\Exceptions\NotConnected
     * @throws \unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined
     */
    public function loop(ClientInterface $client, int $idleMicroseconds = 100000): \Generator
    {
        // First of all: subscribe
        $this->logger->debug('Beginning loop', ['idleMicroseconds' => $idleMicroseconds]);
        $client->setBlocking(true);
        $readableContent = $client->sendData($this);
        // Set blocking explicitly to false due to messages not always arriving in the correct order
        $client->setBlocking(false);

        while (true) {
            $this->logger->debug('--- Loop ---');
            // Only if we receive a Publish event from the broker, yield the contents
            if ($readableContent instanceof Publish) {
                yield $readableContent->getMessage();
            } else {
                // Only wait for a certain amount of time if there was nothing in the queue
                $this->logger->info('Got an incoming object, disregarding', ['class' => \get_class($readableContent)]);
                usleep($idleMicroseconds);
            }

            $readableContent = $this->checkForEvent($client);
        }
    }

    /**
     * A subscription is based on filters, this function allows us to pass on filters
     *
     * @param Topic[] $topics
     * @return Subscribe
     */
    public function addTopics(Topic ...$topics): Subscribe
    {
        $this->topics = $topics;
        $this->logger->debug('Topics added', ['totalTopics', count($this->topics)]);

        return $this;
    }

    /**
     * @param ClientInterface $client
     * @return bool
     */
    protected function checkPingTime(ClientInterface $client): bool
    {
        $this->logger->debug('Checking ping request time');
        if ($client->isItPingTime()) {
            $this->logger->notice('PingReq is needed, sending');
            $client->setBlocking(true);
            $client->sendData(new PingReq($this->logger));
            $client->setBlocking(false);
        }

        return true;
    }
}


1			<?php
2
3			declare(strict_types=1);
4
5			namespace unreal4u\MQTT\Protocol;
6
7			use unreal4u\MQTT\Application\EmptyReadableResponse;
8			use unreal4u\MQTT\Internals\ClientInterface;
9			use unreal4u\MQTT\Internals\EventManager;
10			use unreal4u\MQTT\Internals\ProtocolBase;
11			use unreal4u\MQTT\Internals\ReadableContentInterface;
12			use unreal4u\MQTT\Internals\WritableContent;
13			use unreal4u\MQTT\Internals\WritableContentInterface;
14			use unreal4u\MQTT\Protocol\Subscribe\Topic;
15			use unreal4u\MQTT\Utilities;
16
17			final class Subscribe extends ProtocolBase implements WritableContentInterface
18			{
19			use WritableContent;
20
21			const CONTROL_PACKET_VALUE = 8;
22
23			private $packetIdentifier = 0;
24
25			/**
26			* An array of topics on which to subscribe to
27			* @var Topic[]
28			*/
29			private $topics = [];
30
31			/**
32			* @return string
33			* @throws \OutOfRangeException
34			* @throws \Exception
35			*/
36			public function createVariableHeader(): string
37			{
38			// Subscribe must always send a 2 flag
39			$this->specialFlags = 2;
40
41			// Assign a packet identifier automatically if none has been assigned yet
42			if ($this->packetIdentifier === 0) {
43			$this->setPacketIdentifier(random_int(0, 65535));
44			}
45
46			return Utilities::convertNumberToBinaryString($this->packetIdentifier);
47			}
48
49			public function createPayload(): string
50			{
51			$output = '';
52			foreach ($this->topics as $topic) {
53			// chr on QoS level is safe because it will create an 8-bit flag where the first 6 are only 0's
54			$output .= $this->createUTF8String($topic->getTopicName()) . \chr($topic->getTopicQoSLevel());
55			}
56			return $output;
57			}
58
59			/**
60			* When the Server receives a SUBSCRIBE Packet from a Client, the Server MUST respond with a SUBACK Packet
61			*
62			* This can however not be in the same order, as we may be able to receive PUBLISH packets before getting a SUBACK
63			* back
64			*
65			* @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718134 (MQTT-3.8.4-1)
66			* @return bool
67			*/
68			public function shouldExpectAnswer(): bool
69			{
70			return true;
71			}
72
73			/**
74			* SUBSCRIBE Control Packets MUST contain a non-zero 16-bit Packet Identifier
75			*
76			* @param int $packetIdentifier
77			* @return Subscribe
78			* @throws \OutOfRangeException
79			*/
80			public function setPacketIdentifier(int $packetIdentifier): Subscribe
81			{
82			if ($packetIdentifier > 65535 \|\| $packetIdentifier < 1) {
83			throw new \OutOfRangeException('Packet identifier must fit within 2 bytes');
84			}
85
86			$this->packetIdentifier = $packetIdentifier;
87
88			return $this;
89			}
90
91			public function getPacketIdentifier(): int
92			{
93			return $this->packetIdentifier;
94			}
95
96			/**
97			* Performs a check on the socket connection and returns either the contents or an empty object
98			*
99			* @param ClientInterface $client
100			* @return ReadableContentInterface
101			* @throws \DomainException
102			* @throws \unreal4u\MQTT\Exceptions\NotConnected
103			* @throws \unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined
104			*/
105			public function checkForEvent(ClientInterface $client): ReadableContentInterface
106			{
107			$this->checkPingTime($client);
108			$publishPacketControlField = $client->readSocketData(1);
109			$eventManager = new EventManager($this->logger);
110
111			$this->logger->debug('Checking event', ['ordValue' => \ord($publishPacketControlField) & 255]);
112			if ((\ord($publishPacketControlField) & 255) > 0) {
113			return $eventManager->analyzeHeaders($publishPacketControlField, $client);
114			}
115
116			$this->logger->debug('No valid publish packet control field found, returning empty response');
117			return new EmptyReadableResponse($this->logger);
118			}
119
120			/**
121			* Loop and yields different type of results back whenever they are available
122			*
123			* @param ClientInterface $client
124			* @param int $idleMicroseconds The amount of microseconds the watcher should wait before checking the socket again
125			* @return \Generator
126			* @throws \DomainException
127			* @throws \unreal4u\MQTT\Exceptions\NotConnected
128			* @throws \unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined
129			*/
130			public function loop(ClientInterface $client, int $idleMicroseconds = 100000): \Generator
131			{
132			// First of all: subscribe
133			$this->logger->debug('Beginning loop', ['idleMicroseconds' => $idleMicroseconds]);
134			$client->setBlocking(true);
135			$readableContent = $client->sendData($this);
136			// Set blocking explicitly to false due to messages not always arriving in the correct order
137			$client->setBlocking(false);
138
139			while (true) {
140			$this->logger->debug('--- Loop ---');
141			// Only if we receive a Publish event from the broker, yield the contents
142			if ($readableContent instanceof Publish) {
143			yield $readableContent->getMessage();
144			} else {
145			// Only wait for a certain amount of time if there was nothing in the queue
146			$this->logger->info('Got an incoming object, disregarding', ['class' => \get_class($readableContent)]);
147			usleep($idleMicroseconds);
148			}
149
150			$readableContent = $this->checkForEvent($client);
151			}
152			}
153
154			/**
155			* A subscription is based on filters, this function allows us to pass on filters
156			*
157			* @param Topic[] $topics
158			* @return Subscribe
159			*/
160			public function addTopics(Topic ...$topics): Subscribe
161			{
162			$this->topics = $topics;
163			$this->logger->debug('Topics added', ['totalTopics', count($this->topics)]);
164
165			return $this;
166			}
167
168			/**
169			* @param ClientInterface $client
170			* @return bool
171			*/
172			protected function checkPingTime(ClientInterface $client): bool
173			{
174			$this->logger->debug('Checking ping request time');
175			if ($client->isItPingTime()) {
176			$this->logger->notice('PingReq is needed, sending');
177			$client->setBlocking(true);
178			$client->sendData(new PingReq($this->logger));
179			$client->setBlocking(false);
180			}
181
182			return true;
183			}
184			}
185

unreal4u / mqtt

Push — master ( c8a884...c0f80b )

Subscribe::loop() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like