Publish::createVariableHeader() - Code Metrics - Inspection of "First steps in QoS level 2 message reception from..." - unreal4u/mqtt - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 474237...d94a2b )

by Camilo

created 2018-02-03 21:49 UTC

Publish::createVariableHeader() B

↳ Parent: Publish

Complexity

Conditions	5
Paths	9

Size

Total Lines	35
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	16
CRAP Score	5.0342

Importance

Changes

Metric	Value
dl	0
loc	35
ccs	16
cts	18
cp	0.8889
rs	8.439
c	0
b	0
f	0
cc	5
eloc	17
nc	9
nop	0
crap	5.0342

<?php

declare(strict_types=1);

namespace unreal4u\MQTT\Protocol;

use unreal4u\MQTT\Application\EmptyReadableResponse;
use unreal4u\MQTT\Application\Message;
use unreal4u\MQTT\Application\Topic;
use unreal4u\MQTT\DataTypes\QoSLevel;
use unreal4u\MQTT\Internals\ClientInterface;
use unreal4u\MQTT\Internals\ProtocolBase;
use unreal4u\MQTT\Internals\ReadableContent;
use unreal4u\MQTT\Internals\ReadableContentInterface;
use unreal4u\MQTT\Internals\WritableContent;
use unreal4u\MQTT\Internals\WritableContentInterface;
use unreal4u\MQTT\Utilities;

/**
 * A PUBLISH Control Packet is sent from a Client to a Server or vice-versa to transport an Application Message.
 *
 * @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718037
 */
final class Publish extends ProtocolBase implements ReadableContentInterface, WritableContentInterface
{
    use ReadableContent, WritableContent;

    const CONTROL_PACKET_VALUE = 3;

    /**
     * Contains the message to be sent
     * @var Message
     */
    private $message;

    /**
     * Some interchanges with the broker will send or receive a packet identifier
     * @var int
     */
    public $packetIdentifier = 0;

    /**
     * Flag to check whether a message is a redelivery (DUP flag)
     * @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718038
     * @var bool
     */
    public $isRedelivery = false;

    /**
     * @return string
     * @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
     * @throws \unreal4u\MQTT\Exceptions\MissingTopicName
     * @throws \OutOfRangeException
     * @throws \InvalidArgumentException
     */
    public function createVariableHeader(): string
    {
        if ($this->message === null) {
            throw new \InvalidArgumentException('You must at least provide a message object with a topic name');
        }

        $bitString = $this->createUTF8String($this->message->getTopicName());
        // Reset the special flags should the object be reused with another message
        $this->specialFlags = 0;

        if ($this->isRedelivery) {
            $this->logger->debug('Activating redelivery bit');
            // DUP flag: if the message is a re-delivery, mark it as such
            $this->specialFlags |= 8;
        }

        // Check QoS level and perform the corresponding actions
        if ($this->message->getQoSLevel() !== 0) {
            // 0 for QoS lvl2 for QoS lvl1 and 4 for QoS lvl2
            $this->specialFlags |= ($this->message->getQoSLevel() * 2);
            $bitString .= Utilities::convertNumberToBinaryString($this->packetIdentifier);
            $this->logger->debug(sprintf('Activating QoS level %d bit', $this->message->getQoSLevel()), [
                'specialFlags' => $this->specialFlags,
            ]);
        }

        if ($this->message->isRetained()) {
            // RETAIN flag: should the server retain the message?
            $this->specialFlags |= 1;
            $this->logger->debug('Activating retain flag', ['specialFlags' => $this->specialFlags]);
        }

        $this->logger->info('Variable header created', ['specialFlags' => $this->specialFlags]);

        return $bitString;
    }

    public function createPayload(): string
    {
        if (!$this->message->validateMessage()) {
            throw new \InvalidArgumentException('Invalid message');
        }

        return $this->message->getPayload();
    }

    /**
     * QoS level 0 does not have to wait for a answer, so return false. Any other QoS level returns true
     * @return bool
     * @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
     */
    public function shouldExpectAnswer(): bool
    {
        $shouldExpectAnswer = !($this->message->getQoSLevel() === 0);
        $this->logger->debug('Checking whether we should expect an answer or not', [
            'shouldExpectAnswer' => $shouldExpectAnswer,
        ]);
        return $shouldExpectAnswer;
    }

    public function expectAnswer(string $data, ClientInterface $client): ReadableContentInterface
    {
        switch ($this->message->getQoSLevel()) {
            case 1:
                $pubAck = new PubAck($this->logger);
                $pubAck->instantiateObject($data, $client);
                return $pubAck;
            case 2:
                $pubRec = new PubRec($this->logger);
                $pubRec->instantiateObject($data, $client);
                return $pubRec;
            case 0:
            default:
                return new EmptyReadableResponse($this->logger);
        }
    }

    /**
     * Sets the to be sent message
     *
     * @param Message $message
     * @return WritableContentInterface
     */
    public function setMessage(Message $message): WritableContentInterface
    {
        $this->message = $message;
        return $this;
    }

    /**
     * Gets the set message
     *
     * @return Message
     */
    public function getMessage(): Message
    {
        return $this->message;
    }

    /**
     * Sets several bits and pieces from the first byte of the fixed header for the Publish packet
     *
     * @param int $firstByte
     * @return Publish
     * @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
     */
    private function analyzeFirstByte(int $firstByte): Publish
    {
        $this->logger->debug('Analyzing first byte', [sprintf('%08d', decbin($firstByte))]);
        // Retained bit is bit 0 of first byte
        $this->message->setRetainFlag(false);
        if ($firstByte & 1) {
            $this->logger->debug('Setting retain flag to true');
            $this->message->setRetainFlag(true);
        }
        // QoS level are the last bits 2 & 1 of the first byte
        $this->message->setQoSLevel($this->determineIncomingQoSLevel($firstByte));

        // Duplicate message must be checked only on QoS > 0, else set it to false
        $this->isRedelivery = false;
        if ($firstByte & 8 && $this->message->getQoSLevel() !== 0) {
            // Is a duplicate is always bit 3 of first byte
            $this->isRedelivery = true;
            $this->logger->debug('Setting redelivery bit');
        }

        return $this;
    }

    /**
     * Finds out the QoS level in a fixed header for the Publish object
     *
     * @param int $bitString
     * @return QoSLevel
     * @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
     */
    private function determineIncomingQoSLevel(int $bitString): QoSLevel
    {
        // Strange operation, why? Because 4 == QoS lvl2; 2 == QoS lvl1, 0 == QoS lvl0
        $incomingQoSLevel = ($bitString & 4) / 2;
        $this->logger->debug('Setting QoS level', ['incomingQoSLevel' => $incomingQoSLevel]);
        return new QoSLevel($incomingQoSLevel);
    }

    /**
     * Will perform sanity checks and fill in the Readable object with data
     * @param string $rawMQTTHeaders
     * @param ClientInterface $client
     * @return ReadableContentInterface
     * @throws \OutOfBoundsException
     * @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
     * @throws \InvalidArgumentException
     * @throws \OutOfRangeException
     */
    public function fillObject(string $rawMQTTHeaders, ClientInterface $client): ReadableContentInterface
    {
        if (\strlen($rawMQTTHeaders) === 1) {
            $this->logger->debug('Fast check, read rest of data from socket');
            $restOfBytes = $client->readSocketData(1);
            $payload = $client->readSocketData(\ord($restOfBytes));
        } else {
            $this->logger->debug('Slow form, retransform data and read rest of data');
            $restOfBytes = $rawMQTTHeaders{1};
            $payload = substr($rawMQTTHeaders, 2);
            $exactRest = \ord($restOfBytes) - \strlen($payload);
            $payload .= $client->readSocketData($exactRest);
            $rawMQTTHeaders = $rawMQTTHeaders{0};
        }

        // At this point, $rawMQTTHeaders will be always 1 byte long
        $this->message = new Message();
        $this->analyzeFirstByte(\ord($rawMQTTHeaders));
        // $rawMQTTHeaders may be redefined
        $rawMQTTHeaders = $rawMQTTHeaders . $restOfBytes . $payload;

        // Topic size is always the 3rd byte
        $topicSize = \ord($rawMQTTHeaders{3});

        $messageStartPosition = 4;
        if ($this->message->getQoSLevel() > 0) {
            // [2 (fixed header) + 2 (topic size) + $topicSize] marks the beginning of the 2 packet identifier bytes
            $this->packetIdentifier = Utilities::convertBinaryStringToNumber(
                $rawMQTTHeaders{5 + $topicSize} . $rawMQTTHeaders{4 + $topicSize}
            );
            $messageStartPosition += 2;
        }

        $this->logger->debug('Determined headers', [
            'topicSize' => $topicSize,
            'QoSLevel' => $this->message->getQoSLevel(),
            'isDuplicate' => $this->isRedelivery,
            'isRetained' => $this->message->isRetained(),
            'packetIdentifier' => $this->packetIdentifier,
        ]);

        $this->message->setPayload(substr($rawMQTTHeaders, $messageStartPosition + $topicSize));
        $this->message->setTopic(new Topic(substr($rawMQTTHeaders, $messageStartPosition, $topicSize)));

        return $this;
    }

    /**
     * @inheritdoc
     * @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
     * @throws \unreal4u\MQTT\Exceptions\ServerClosedConnection
     * @throws \unreal4u\MQTT\Exceptions\NotConnected
     * @throws \unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined
     */
    public function performSpecialActions(ClientInterface $client, WritableContentInterface $originalRequest): bool
    {
        $qosLevel = $this->message->getQoSLevel();
        if ($qosLevel === 0) {
            $this->logger->debug('No response needed', ['qosLevel', $qosLevel]);
        } else {
            if ($qosLevel === 1) {
                $this->logger->debug('Responding with PubAck', ['qosLevel' => $qosLevel]);
                $client->sendData($this->composePubAckAnswer());
            } elseif ($qosLevel === 2) {
                $this->logger->debug('Responding with PubRec', ['qosLevel' => $qosLevel]);
                $client->sendData($this->composePubRecAnswer());
            }
        }

        return true;
    }

    private function composePubRecAnswer(): PubRec
    {
        $pubRec = new PubRec($this->logger);
        $pubRec->packetIdentifier = $this->packetIdentifier;
        return $pubRec;
    }

    /**
     * Composes a PubAck answer with the same packetIdentifier as what we received
     * @return PubAck
     */
    private function composePubAckAnswer(): PubAck
    {
        $pubAck = new PubAck($this->logger);
        $pubAck->packetIdentifier = $this->packetIdentifier;
        return $pubAck;
    }
}


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\Application\Message;
9		use unreal4u\MQTT\Application\Topic;
10		use unreal4u\MQTT\DataTypes\QoSLevel;
11		use unreal4u\MQTT\Internals\ClientInterface;
12		use unreal4u\MQTT\Internals\ProtocolBase;
13		use unreal4u\MQTT\Internals\ReadableContent;
14		use unreal4u\MQTT\Internals\ReadableContentInterface;
15		use unreal4u\MQTT\Internals\WritableContent;
16		use unreal4u\MQTT\Internals\WritableContentInterface;
17		use unreal4u\MQTT\Utilities;
18
19		/**
20		* A PUBLISH Control Packet is sent from a Client to a Server or vice-versa to transport an Application Message.
21		*
22		* @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718037
23		*/
24		final class Publish extends ProtocolBase implements ReadableContentInterface, WritableContentInterface
25		{
26		use ReadableContent, WritableContent;
27
28		const CONTROL_PACKET_VALUE = 3;
29
30		/**
31		* Contains the message to be sent
32		* @var Message
33		*/
34		private $message;
35
36		/**
37		* Some interchanges with the broker will send or receive a packet identifier
38		* @var int
39		*/
40		public $packetIdentifier = 0;
41
42		/**
43		* Flag to check whether a message is a redelivery (DUP flag)
44		* @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718038
45		* @var bool
46		*/
47		public $isRedelivery = false;
48
49		/**
50		* @return string
51		* @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
52		* @throws \unreal4u\MQTT\Exceptions\MissingTopicName
53		* @throws \OutOfRangeException
54		* @throws \InvalidArgumentException
55		*/
56	4	public function createVariableHeader(): string
57		{
58	4	if ($this->message === null) {
59	1	throw new \InvalidArgumentException('You must at least provide a message object with a topic name');
60		}
61
62	3	$bitString = $this->createUTF8String($this->message->getTopicName());
63		// Reset the special flags should the object be reused with another message
64	3	$this->specialFlags = 0;
65
66	3	if ($this->isRedelivery) {
67		$this->logger->debug('Activating redelivery bit');
68		// DUP flag: if the message is a re-delivery, mark it as such
69		$this->specialFlags \|= 8;
70		}
71
72		// Check QoS level and perform the corresponding actions
73	3	if ($this->message->getQoSLevel() !== 0) {
74		// 0 for QoS lvl2 for QoS lvl1 and 4 for QoS lvl2
75	2	$this->specialFlags \|= ($this->message->getQoSLevel() * 2);
76	2	$bitString .= Utilities::convertNumberToBinaryString($this->packetIdentifier);
77	2	$this->logger->debug(sprintf('Activating QoS level %d bit', $this->message->getQoSLevel()), [
78	2	'specialFlags' => $this->specialFlags,
79		]);
80		}
81
82	3	if ($this->message->isRetained()) {
83		// RETAIN flag: should the server retain the message?
84	1	$this->specialFlags \|= 1;
85	1	$this->logger->debug('Activating retain flag', ['specialFlags' => $this->specialFlags]);
86		}
87
88	3	$this->logger->info('Variable header created', ['specialFlags' => $this->specialFlags]);
89
90	3	return $bitString;
91		}
92
93		public function createPayload(): string
94		{
95		if (!$this->message->validateMessage()) {
96		throw new \InvalidArgumentException('Invalid message');
97		}
98
99		return $this->message->getPayload();
100		}
101
102		/**
103		* QoS level 0 does not have to wait for a answer, so return false. Any other QoS level returns true
104		* @return bool
105		* @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
106		*/
107	2	public function shouldExpectAnswer(): bool
108		{
109	2	$shouldExpectAnswer = !($this->message->getQoSLevel() === 0);
110	2	$this->logger->debug('Checking whether we should expect an answer or not', [
111	2	'shouldExpectAnswer' => $shouldExpectAnswer,
112		]);
113	2	return $shouldExpectAnswer;
114		}
115
116	2	public function expectAnswer(string $data, ClientInterface $client): ReadableContentInterface
117		{
118	2	switch ($this->message->getQoSLevel()) {
119	2	case 1:
120	1	$pubAck = new PubAck($this->logger);
121	1	$pubAck->instantiateObject($data, $client);
122	1	return $pubAck;
123	1	case 2:
124		$pubRec = new PubRec($this->logger);
125		$pubRec->instantiateObject($data, $client);
126		return $pubRec;
127	1	case 0:
128		default:
129	1	return new EmptyReadableResponse($this->logger);
130		}
131		}
132
133		/**
134		* Sets the to be sent message
135		*
136		* @param Message $message
137		* @return WritableContentInterface
138		*/
139	6	public function setMessage(Message $message): WritableContentInterface
140		{
141	6	$this->message = $message;
142	6	return $this;
143		}
144
145		/**
146		* Gets the set message
147		*
148		* @return Message
149		*/
150		public function getMessage(): Message
151		{
152		return $this->message;
153		}
154
155		/**
156		* Sets several bits and pieces from the first byte of the fixed header for the Publish packet
157		*
158		* @param int $firstByte
159		* @return Publish
160		* @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
161		*/
162		private function analyzeFirstByte(int $firstByte): Publish
163		{
164		$this->logger->debug('Analyzing first byte', [sprintf('%08d', decbin($firstByte))]);
165		// Retained bit is bit 0 of first byte
166		$this->message->setRetainFlag(false);
167		if ($firstByte & 1) {
168		$this->logger->debug('Setting retain flag to true');
169		$this->message->setRetainFlag(true);
170		}
171		// QoS level are the last bits 2 & 1 of the first byte
172		$this->message->setQoSLevel($this->determineIncomingQoSLevel($firstByte));
173
174		// Duplicate message must be checked only on QoS > 0, else set it to false
175		$this->isRedelivery = false;
176		if ($firstByte & 8 && $this->message->getQoSLevel() !== 0) {
177		// Is a duplicate is always bit 3 of first byte
178		$this->isRedelivery = true;
179		$this->logger->debug('Setting redelivery bit');
180		}
181
182		return $this;
183		}
184
185		/**
186		* Finds out the QoS level in a fixed header for the Publish object
187		*
188		* @param int $bitString
189		* @return QoSLevel
190		* @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
191		*/
192		private function determineIncomingQoSLevel(int $bitString): QoSLevel
193		{
194		// Strange operation, why? Because 4 == QoS lvl2; 2 == QoS lvl1, 0 == QoS lvl0
195		$incomingQoSLevel = ($bitString & 4) / 2;
196		$this->logger->debug('Setting QoS level', ['incomingQoSLevel' => $incomingQoSLevel]);
197		return new QoSLevel($incomingQoSLevel);
198		}
199
200		/**
201		* Will perform sanity checks and fill in the Readable object with data
202		* @param string $rawMQTTHeaders
203		* @param ClientInterface $client
204		* @return ReadableContentInterface
205		* @throws \OutOfBoundsException
206		* @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
207		* @throws \InvalidArgumentException
208		* @throws \OutOfRangeException
209		*/
210		public function fillObject(string $rawMQTTHeaders, ClientInterface $client): ReadableContentInterface
211		{
212		if (\strlen($rawMQTTHeaders) === 1) {
213		$this->logger->debug('Fast check, read rest of data from socket');
214		$restOfBytes = $client->readSocketData(1);
215		$payload = $client->readSocketData(\ord($restOfBytes));
216		} else {
217		$this->logger->debug('Slow form, retransform data and read rest of data');
218		$restOfBytes = $rawMQTTHeaders{1};
219		$payload = substr($rawMQTTHeaders, 2);
220		$exactRest = \ord($restOfBytes) - \strlen($payload);
221		$payload .= $client->readSocketData($exactRest);
222		$rawMQTTHeaders = $rawMQTTHeaders{0};
223		}
224
225		// At this point, $rawMQTTHeaders will be always 1 byte long
226		$this->message = new Message();
227		$this->analyzeFirstByte(\ord($rawMQTTHeaders));
228		// $rawMQTTHeaders may be redefined
229		$rawMQTTHeaders = $rawMQTTHeaders . $restOfBytes . $payload;
230
231		// Topic size is always the 3rd byte
232		$topicSize = \ord($rawMQTTHeaders{3});
233
234		$messageStartPosition = 4;
235		if ($this->message->getQoSLevel() > 0) {
236		// [2 (fixed header) + 2 (topic size) + $topicSize] marks the beginning of the 2 packet identifier bytes
237		$this->packetIdentifier = Utilities::convertBinaryStringToNumber(
238		$rawMQTTHeaders{5 + $topicSize} . $rawMQTTHeaders{4 + $topicSize}
239		);
240		$messageStartPosition += 2;
241		}
242
243		$this->logger->debug('Determined headers', [
244		'topicSize' => $topicSize,
245		'QoSLevel' => $this->message->getQoSLevel(),
246		'isDuplicate' => $this->isRedelivery,
247		'isRetained' => $this->message->isRetained(),
248		'packetIdentifier' => $this->packetIdentifier,
249		]);
250
251		$this->message->setPayload(substr($rawMQTTHeaders, $messageStartPosition + $topicSize));
252		$this->message->setTopic(new Topic(substr($rawMQTTHeaders, $messageStartPosition, $topicSize)));
253
254		return $this;
255		}
256
257		/**
258		* @inheritdoc
259		* @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
260		* @throws \unreal4u\MQTT\Exceptions\ServerClosedConnection
261		* @throws \unreal4u\MQTT\Exceptions\NotConnected
262		* @throws \unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined
263		*/
264		public function performSpecialActions(ClientInterface $client, WritableContentInterface $originalRequest): bool
265		{
266		$qosLevel = $this->message->getQoSLevel();
267		if ($qosLevel === 0) {
268		$this->logger->debug('No response needed', ['qosLevel', $qosLevel]);
269		} else {
270		if ($qosLevel === 1) {
271		$this->logger->debug('Responding with PubAck', ['qosLevel' => $qosLevel]);
272		$client->sendData($this->composePubAckAnswer());
273		} elseif ($qosLevel === 2) {
274		$this->logger->debug('Responding with PubRec', ['qosLevel' => $qosLevel]);
275		$client->sendData($this->composePubRecAnswer());
276		}
277		}
278
279		return true;
280		}
281
282		private function composePubRecAnswer(): PubRec
283		{
284		$pubRec = new PubRec($this->logger);
285		$pubRec->packetIdentifier = $this->packetIdentifier;
286		return $pubRec;
287		}
288
289		/**
290		* Composes a PubAck answer with the same packetIdentifier as what we received
291		* @return PubAck
292		*/
293		private function composePubAckAnswer(): PubAck
294		{
295		$pubAck = new PubAck($this->logger);
296		$pubAck->packetIdentifier = $this->packetIdentifier;
297		return $pubAck;
298		}
299		}
300

unreal4u / mqtt

Push — master ( 474237...d94a2b )

Publish::createVariableHeader() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like