Publish::createVariableHeader() - Code Metrics - Inspection of "Introducing more VO's" - unreal4u/mqtt - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 325ba3...57475a )

by Camilo

created 2018-01-30 17:00 UTC

Publish::createVariableHeader() B

↳ Parent: Publish

Complexity

Conditions	5
Paths	9

Size

Total Lines	36
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	30

Importance

Changes

Metric	Value
dl	0
loc	36
ccs	0
cts	19
cp	0
rs	8.439
c	0
b	0
f	0
cc	5
eloc	18
nc	9
nop	0
crap	30

<?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\Exceptions\InvalidQoSLevel;
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;

    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) {
            // 2 for QoS lvl1 and 4 for QoS lvl2
            $this->specialFlags |= ($this->message->getQoSLevel() * 2);
            $this->packetIdentifier++;
            $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
     */
    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->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;
        }

        return $this;
    }

    /**
     * Finds out the QoS level in a fixed header for the Publish object
     *
     * @param int $bitString
     * @return int
     * @throws \unreal4u\MQTT\Exceptions\InvalidQoSLevel
     */
    private function determineIncomingQoSLevel(int $bitString): int
    {
        // QoS lvl 6 does not exist, throw exception
        if (($bitString & 6) >= 6) {
            throw new InvalidQoSLevel('Invalid QoS level "' . $bitString . '" found (both bits set?)');
        }

        // Strange operation, why? Because 4 == QoS lvl2; 2 == QoS lvl1, 0 == QoS lvl0
        return $bitString & 4 / 2;
    }

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

        return true;
    }

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

unreal4u / mqtt

Push — master ( 325ba3...57475a )

Publish::createVariableHeader() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like