Publish::getMessage() - Code Metrics - Inspection of "Don't use deprecated stuff" - unreal4u/mqtt - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d71bca...578a67 )

by Camilo

created 2020-09-22 07:54 UTC

Publish::getMessage() A

↳ Parent: Publish

Complexity

Conditions	1
Paths	1

Size

Total Lines	3
Code Lines	1

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
eloc	1
c	1
b	0
f	0
dl	0
loc	3
ccs	0
cts	2
cp	0
rs	10
cc	1
nc	1
nop	0
crap	2

<?php

declare(strict_types=1);

namespace unreal4u\MQTT\Protocol;

use InvalidArgumentException;
use OutOfBoundsException;
use OutOfRangeException;
use unreal4u\MQTT\Application\EmptyReadableResponse;
use unreal4u\MQTT\DataTypes\Message;
use unreal4u\MQTT\DataTypes\PacketIdentifier;
use unreal4u\MQTT\DataTypes\QoSLevel;
use unreal4u\MQTT\DataTypes\TopicName;
use unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined;
use unreal4u\MQTT\Exceptions\InvalidQoSLevel;
use unreal4u\MQTT\Exceptions\InvalidRequest;
use unreal4u\MQTT\Exceptions\InvalidResponseType;
use unreal4u\MQTT\Exceptions\MessageTooBig;
use unreal4u\MQTT\Exceptions\MissingTopicName;
use unreal4u\MQTT\Exceptions\NotConnected;
use unreal4u\MQTT\Internals\ClientInterface;
use unreal4u\MQTT\Internals\PacketIdentifierFunctionality;
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;

use function decbin;
use function ord;
use function sprintf;
use function strlen;
use function substr;

/**
 * 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
 *
 * QoS lvl1:
 *   First packet: PUBLISH
 *   Second packet: PUBACK
 *
 * QoS lvl2:
 *   First packet: PUBLISH
 *   Second packet: PUBREC
 *   Third packet: PUBREL
 *   Fourth packet: PUBCOMP
 *
 * @see https://go.gliffy.com/go/publish/12498076
 */
final class Publish extends ProtocolBase implements ReadableContentInterface, WritableContentInterface
{
    use ReadableContent;
    use /** @noinspection TraitsPropertiesConflictsInspection */
        WritableContent;
    use PacketIdentifierFunctionality;

    private const CONTROL_PACKET_VALUE = 3;

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

    /**
     * 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 InvalidQoSLevel
     * @throws 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');
        }

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

        $variableHeaderContents .= $this->createVariableHeaderFlags();

        return $variableHeaderContents;
    }

    /**
     * Sets some common flags and returns the variable header string should there be one
     *
     * @return string
     * @throws OutOfRangeException
     * @throws InvalidQoSLevel
     */
    private function createVariableHeaderFlags(): string
    {
        if ($this->isRedelivery) {
            // DUP flag: if the message is a re-delivery, mark it as such
            $this->specialFlags |= 8;
            $this->logger->debug('Activating redelivery bit');
        }

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

        // 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);
            $this->logger->debug(sprintf('Activating QoS level %d bit', $this->message->getQoSLevel()));
            return $this->getPacketIdentifierBinaryRepresentation();
        }

        return '';
    }

    /**
     * @return string
     * @throws MissingTopicName
     * @throws MessageTooBig
     * @throws InvalidArgumentException
     */
    public function createPayload(): string
    {
        if ($this->message === null) {
            throw new InvalidArgumentException('A message must be set before publishing');
        }
        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 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;
    }

    /**
     * @param string $brokerBitStream
     * @param ClientInterface $client
     * @return ReadableContentInterface
     * @throws InvalidResponseType
     */
    public function expectAnswer(string $brokerBitStream, ClientInterface $client): ReadableContentInterface
    {
        switch ($this->message->getQoSLevel()) {
            case 1:
                $pubAck = new PubAck($this->logger);
                $pubAck->instantiateObject($brokerBitStream, $client);
                return $pubAck;
            case 2:
                $pubRec = new PubRec($this->logger);
                $pubRec->instantiateObject($brokerBitStream, $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
     * @param QoSLevel $qoSLevel
     * @return Publish
     * @throws InvalidQoSLevel
     */
    private function analyzeFirstByte(int $firstByte, QoSLevel $qoSLevel): 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) === 1) {
            $this->logger->debug('Setting retain flag to true');
            $this->message->setRetainFlag(true);
        }
        // QoS level is already been taken care of, assign it to the message at this point
        $this->message->setQoSLevel($qoSLevel);

        // Duplicate message must be checked only on QoS > 0, else set it to false
        $this->isRedelivery = false;
        if (($firstByte & 8) === 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 InvalidQoSLevel
     */
    private function determineIncomingQoSLevel(int $bitString): QoSLevel
    {
        // QoS lvl are in bit positions 1-2. Shifting is strictly speaking not needed, but increases human comprehension
        $shiftedBits = $bitString >> 1;
        $incomingQoSLevel = 0;
        if (($shiftedBits & 1) === 1) {
            $incomingQoSLevel = 1;
        }
        if (($shiftedBits & 2) === 2) {
            $incomingQoSLevel = 2;
        }

        $this->logger->debug('Setting QoS level', ['bitString' => $bitString, 'incomingQoSLevel' => $incomingQoSLevel]);
        return new QoSLevel($incomingQoSLevel);
    }

    /**
     * Gets the full message in case this object needs to
     *
     * @param string $rawMQTTHeaders
     * @param ClientInterface $client
     * @return string
     * @throws OutOfBoundsException
     * @throws InvalidArgumentException
     * @throws MessageTooBig
     * @throws InvalidQoSLevel
     */
    private function completePossibleIncompleteMessage(string $rawMQTTHeaders, ClientInterface $client): string
    {
        // Read at least one extra byte from the stream if we know that the message is too short
        if (strlen($rawMQTTHeaders) < 2) {
            $rawMQTTHeaders .= $client->readBrokerData(1);
        }

        $restOfBytes = $this->performRemainingLengthFieldOperations($rawMQTTHeaders, $client);

        /*
         * A complete message consists of:
         *  - The very first byte
         *  - The size of the remaining length field (from 1 to 4 bytes)
         *  - The $restOfBytes
         *
         * So we have to compare what we already have vs the above calculation
         *
         * More information:
         * http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html#_Toc442180832
         */
        if (strlen($rawMQTTHeaders) < ($restOfBytes + $this->sizeOfRemainingLengthField + 1)) {
            // Read only the portion of data we have left from the socket
            $readableDataLeft = ($restOfBytes + $this->sizeOfRemainingLengthField + 1) - strlen($rawMQTTHeaders);
            $rawMQTTHeaders .= $client->readBrokerData($readableDataLeft);
        }

        return $rawMQTTHeaders;
    }

    /**
     * Will perform sanity checks and fill in the Readable object with data
     * @param string $rawMQTTHeaders
     * @param ClientInterface $client
     * @return ReadableContentInterface
     * @throws MessageTooBig
     * @throws OutOfBoundsException
     * @throws InvalidQoSLevel
     * @throws InvalidArgumentException
     * @throws OutOfRangeException
     */
    public function fillObject(string $rawMQTTHeaders, ClientInterface $client): ReadableContentInterface
    {
        // Retrieve full message first
        $fullMessage = $this->completePossibleIncompleteMessage($rawMQTTHeaders, $client);
        // Handy to maintain for debugging purposes
        #$this->logger->debug('Bin data', [\unreal4u\MQTT\DebugTools::convertToBinaryRepresentation($rawMQTTHeaders)]);

        // Handy to have: the first byte
        $firstByte = ord($fullMessage{0});
        // TopicName size is always on the second position after the size of the remaining length field (1 to 4 bytes)
        $topicSize = ord($fullMessage{$this->sizeOfRemainingLengthField + 2});
        // With the first byte, we can determine the QoS level of the incoming message
        $qosLevel = $this->determineIncomingQoSLevel($firstByte);

        $messageStartPosition = $this->sizeOfRemainingLengthField + 3;
        // If we have a QoS level present, we must retrieve the packet identifier as well
        if ($qosLevel->getQoSLevel() > 0) {
            $this->logger->debug('QoS level above 0, shifting message start position and getting packet identifier');
            // [2 (fixed header) + 2 (topic size) + $topicSize] marks the beginning of the 2 packet identifier bytes
            $this->setPacketIdentifier(new PacketIdentifier(Utilities::convertBinaryStringToNumber(
                $fullMessage{$this->sizeOfRemainingLengthField + 3 + $topicSize} .
                $fullMessage{$this->sizeOfRemainingLengthField + 4 + $topicSize}
            )));
            $this->logger->debug('Determined packet identifier', ['PI' => $this->getPacketIdentifier()]);
            $messageStartPosition += 2;
        }

        // At this point $rawMQTTHeaders will be always 1 byte long, initialize a Message object with dummy data for now
        $this->message = new Message(
            // Save to assume a constant here: first 2 bytes will always be fixed header, next 2 bytes are topic size
            substr($fullMessage, $messageStartPosition + $topicSize),
            new TopicName(substr($fullMessage, $this->sizeOfRemainingLengthField + 3, $topicSize))
        );
        $this->analyzeFirstByte($firstByte, $qosLevel);

        $this->logger->debug('Determined headers', [
            'topicSize' => $topicSize,
            'QoSLevel' => $this->message->getQoSLevel(),
            'isDuplicate' => $this->isRedelivery,
            'isRetained' => $this->message->isRetained(),
            #'packetIdentifier' => $this->packetIdentifier->getPacketIdentifierValue(), // This is not always set!
        ]);

        return $this;
    }

    /**
     * @inheritdoc
     * @throws InvalidRequest
     * @throws InvalidQoSLevel
     * @throws NotConnected
     * @throws 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->processObject($this->composePubAckAnswer());
            } elseif ($qosLevel === 2) {
                $this->logger->debug('Responding with PubRec', ['qosLevel' => $qosLevel]);
                $client->processObject($this->composePubRecAnswer());
            }
        }

        return true;
    }

    /**
     * Composes a PubRec answer with the same packetIdentifier as what we received
     *
     * @return PubRec
     * @throws InvalidRequest
     */
    private function composePubRecAnswer(): PubRec
    {
        $this->checkForValidPacketIdentifier();
        $pubRec = new PubRec($this->logger);
        $pubRec->setPacketIdentifier($this->packetIdentifier);
        return $pubRec;
    }

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

    /**
     * Will check whether the current object has a packet identifier set. If not, we are in serious problems!
     *
     * @return Publish
     * @throws InvalidRequest
     */
    private function checkForValidPacketIdentifier(): self
    {
        if ($this->packetIdentifier === null) {
            $this->logger->critical('No valid packet identifier found at a stage where there MUST be one set');
            throw new InvalidRequest('You are trying to send a request without a valid packet identifier');
        }

        return $this;
    }

    /**
     * PUBLISH packet is the exception to the rule: it is not started on base of a packet that gets sent by us
     */
    public function getOriginControlPacket(): int
    {
        return 0;
    }
}


1			<?php
2
3			declare(strict_types=1);
4
5			namespace unreal4u\MQTT\Protocol;
6
7			use InvalidArgumentException;
8			use OutOfBoundsException;
9			use OutOfRangeException;
10			use unreal4u\MQTT\Application\EmptyReadableResponse;
11			use unreal4u\MQTT\DataTypes\Message;
12			use unreal4u\MQTT\DataTypes\PacketIdentifier;
13			use unreal4u\MQTT\DataTypes\QoSLevel;
14			use unreal4u\MQTT\DataTypes\TopicName;
15			use unreal4u\MQTT\Exceptions\Connect\NoConnectionParametersDefined;
16			use unreal4u\MQTT\Exceptions\InvalidQoSLevel;
17			use unreal4u\MQTT\Exceptions\InvalidRequest;
18			use unreal4u\MQTT\Exceptions\InvalidResponseType;
19			use unreal4u\MQTT\Exceptions\MessageTooBig;
20			use unreal4u\MQTT\Exceptions\MissingTopicName;
21			use unreal4u\MQTT\Exceptions\NotConnected;
22			use unreal4u\MQTT\Internals\ClientInterface;
23			use unreal4u\MQTT\Internals\PacketIdentifierFunctionality;
24			use unreal4u\MQTT\Internals\ProtocolBase;
25			use unreal4u\MQTT\Internals\ReadableContent;
26			use unreal4u\MQTT\Internals\ReadableContentInterface;
27			use unreal4u\MQTT\Internals\WritableContent;
28			use unreal4u\MQTT\Internals\WritableContentInterface;
29			use unreal4u\MQTT\Utilities;
30
31			use function decbin;
32			use function ord;
33			use function sprintf;
34			use function strlen;
35			use function substr;
36
37			/**
38			* A PUBLISH Control Packet is sent from a Client to a Server or vice-versa to transport an Application Message.
39			*
40			* @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718037
41			*
42			* QoS lvl1:
43			* First packet: PUBLISH
44			* Second packet: PUBACK
45			*
46			* QoS lvl2:
47			* First packet: PUBLISH
48			* Second packet: PUBREC
49			* Third packet: PUBREL
50			* Fourth packet: PUBCOMP
51			*
52			* @see https://go.gliffy.com/go/publish/12498076
53			*/
54			final class Publish extends ProtocolBase implements ReadableContentInterface, WritableContentInterface
55			{
56			use ReadableContent;
57			use /** @noinspection TraitsPropertiesConflictsInspection */
58			WritableContent;
59			use PacketIdentifierFunctionality;
60
61			private const CONTROL_PACKET_VALUE = 3;
62
63			/**
64			* Contains the message to be sent
65			* @var Message
66			*/
67			private $message;
68
69			/**
70			* Flag to check whether a message is a redelivery (DUP flag)
71			* @see http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718038
72			* @var bool
73			*/
74			public $isRedelivery = false;
75
76			/**
77			* @return string
78			* @throws InvalidQoSLevel
79			* @throws MissingTopicName
80			* @throws OutOfRangeException
81			* @throws InvalidArgumentException
82			*/
83			public function createVariableHeader(): string
84			{
85			if ($this->message === null) {
86			throw new InvalidArgumentException('You must at least provide a message object with a topic name');
87			}
88
89			$variableHeaderContents = $this->createUTF8String($this->message->getTopicName());
90			// Reset the special flags should the same object be reused with another message
91			$this->specialFlags = 0;
92
93			$variableHeaderContents .= $this->createVariableHeaderFlags();
94
95			return $variableHeaderContents;
96			}
97
98			/**
99			* Sets some common flags and returns the variable header string should there be one
100			*
101			* @return string
102			* @throws OutOfRangeException
103			* @throws InvalidQoSLevel
104			*/
105			private function createVariableHeaderFlags(): string
106			{
107			if ($this->isRedelivery) {
108			// DUP flag: if the message is a re-delivery, mark it as such
109			$this->specialFlags \|= 8;
110			$this->logger->debug('Activating redelivery bit');
111			}
112
113			if ($this->message->isRetained()) {
114			// RETAIN flag: should the server retain the message?
115			$this->specialFlags \|= 1;
116			$this->logger->debug('Activating retain flag');
117			}
118
119			// Check QoS level and perform the corresponding actions
120			if ($this->message->getQoSLevel() !== 0) {
121			// 0 for QoS lvl2 for QoS lvl1 and 4 for QoS lvl2
122			$this->specialFlags \|= ($this->message->getQoSLevel() * 2);
123			$this->logger->debug(sprintf('Activating QoS level %d bit', $this->message->getQoSLevel()));
124			return $this->getPacketIdentifierBinaryRepresentation();
125			}
126
127			return '';
128			}
129
130			/**
131			* @return string
132			* @throws MissingTopicName
133			* @throws MessageTooBig
134			* @throws InvalidArgumentException
135			*/
136			public function createPayload(): string
137			{
138			if ($this->message === null) {
139			throw new InvalidArgumentException('A message must be set before publishing');
140			}
141			return $this->message->getPayload();
142			}
143
144			/**
145			* QoS level 0 does not have to wait for a answer, so return false. Any other QoS level returns true
146			* @return bool
147			* @throws InvalidQoSLevel
148			*/
149			public function shouldExpectAnswer(): bool
150			{
151			$shouldExpectAnswer = !($this->message->getQoSLevel() === 0);
152			$this->logger->debug('Checking whether we should expect an answer or not', [
153			'shouldExpectAnswer' => $shouldExpectAnswer,
154			]);
155			return $shouldExpectAnswer;
156			}
157
158			/**
159			* @param string $brokerBitStream
160			* @param ClientInterface $client
161			* @return ReadableContentInterface
162			* @throws InvalidResponseType
163			*/
164			public function expectAnswer(string $brokerBitStream, ClientInterface $client): ReadableContentInterface
165			{
166			switch ($this->message->getQoSLevel()) {
167			case 1:
168			$pubAck = new PubAck($this->logger);
169			$pubAck->instantiateObject($brokerBitStream, $client);
170			return $pubAck;
171			case 2:
172			$pubRec = new PubRec($this->logger);
173			$pubRec->instantiateObject($brokerBitStream, $client);
174			return $pubRec;
175			case 0:
176			default:
177			return new EmptyReadableResponse($this->logger);
178			}
179			}
180
181			/**
182			* Sets the to be sent message
183			*
184			* @param Message $message
185			* @return WritableContentInterface
186			*/
187			public function setMessage(Message $message): WritableContentInterface
188			{
189			$this->message = $message;
190			return $this;
191			}
192
193			/**
194			* Gets the set message
195			*
196			* @return Message
197			*/
198			public function getMessage(): Message
199			{
200			return $this->message;
201			}
202
203			/**
204			* Sets several bits and pieces from the first byte of the fixed header for the Publish packet
205			*
206			* @param int $firstByte
207			* @param QoSLevel $qoSLevel
208			* @return Publish
209			* @throws InvalidQoSLevel
210			*/
211			private function analyzeFirstByte(int $firstByte, QoSLevel $qoSLevel): Publish
212			{
213			$this->logger->debug('Analyzing first byte', [sprintf('%08d', decbin($firstByte))]);
214			// Retained bit is bit 0 of first byte
215			$this->message->setRetainFlag(false);
216			if (($firstByte & 1) === 1) {
217			$this->logger->debug('Setting retain flag to true');
218			$this->message->setRetainFlag(true);
219			}
220			// QoS level is already been taken care of, assign it to the message at this point
221			$this->message->setQoSLevel($qoSLevel);
222
223			// Duplicate message must be checked only on QoS > 0, else set it to false
224			$this->isRedelivery = false;
225			if (($firstByte & 8) === 8 && $this->message->getQoSLevel() !== 0) {
226			// Is a duplicate is always bit 3 of first byte
227			$this->isRedelivery = true;
228			$this->logger->debug('Setting redelivery bit');
229			}
230
231			return $this;
232			}
233
234			/**
235			* Finds out the QoS level in a fixed header for the Publish object
236			*
237			* @param int $bitString
238			* @return QoSLevel
239			* @throws InvalidQoSLevel
240			*/
241			private function determineIncomingQoSLevel(int $bitString): QoSLevel
242			{
243			// QoS lvl are in bit positions 1-2. Shifting is strictly speaking not needed, but increases human comprehension
244			$shiftedBits = $bitString >> 1;
245			$incomingQoSLevel = 0;
246			if (($shiftedBits & 1) === 1) {
247			$incomingQoSLevel = 1;
248			}
249			if (($shiftedBits & 2) === 2) {
250			$incomingQoSLevel = 2;
251			}
252
253			$this->logger->debug('Setting QoS level', ['bitString' => $bitString, 'incomingQoSLevel' => $incomingQoSLevel]);
254			return new QoSLevel($incomingQoSLevel);
255			}
256
257			/**
258			* Gets the full message in case this object needs to
259			*
260			* @param string $rawMQTTHeaders
261			* @param ClientInterface $client
262			* @return string
263			* @throws OutOfBoundsException
264			* @throws InvalidArgumentException
265			* @throws MessageTooBig
266			* @throws InvalidQoSLevel
267			*/
268			private function completePossibleIncompleteMessage(string $rawMQTTHeaders, ClientInterface $client): string
269			{
270			// Read at least one extra byte from the stream if we know that the message is too short
271			if (strlen($rawMQTTHeaders) < 2) {
272			$rawMQTTHeaders .= $client->readBrokerData(1);
273			}
274
275			$restOfBytes = $this->performRemainingLengthFieldOperations($rawMQTTHeaders, $client);
276
277			/*
278			* A complete message consists of:
279			* - The very first byte
280			* - The size of the remaining length field (from 1 to 4 bytes)
281			* - The $restOfBytes
282			*
283			* So we have to compare what we already have vs the above calculation
284			*
285			* More information:
286			* http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html#_Toc442180832
287			*/
288			if (strlen($rawMQTTHeaders) < ($restOfBytes + $this->sizeOfRemainingLengthField + 1)) {
289			// Read only the portion of data we have left from the socket
290			$readableDataLeft = ($restOfBytes + $this->sizeOfRemainingLengthField + 1) - strlen($rawMQTTHeaders);
291			$rawMQTTHeaders .= $client->readBrokerData($readableDataLeft);
292			}
293
294			return $rawMQTTHeaders;
295			}
296
297			/**
298			* Will perform sanity checks and fill in the Readable object with data
299			* @param string $rawMQTTHeaders
300			* @param ClientInterface $client
301			* @return ReadableContentInterface
302			* @throws MessageTooBig
303			* @throws OutOfBoundsException
304			* @throws InvalidQoSLevel
305			* @throws InvalidArgumentException
306			* @throws OutOfRangeException
307			*/
308			public function fillObject(string $rawMQTTHeaders, ClientInterface $client): ReadableContentInterface
309			{
310			// Retrieve full message first
311			$fullMessage = $this->completePossibleIncompleteMessage($rawMQTTHeaders, $client);
312			// Handy to maintain for debugging purposes
313			#$this->logger->debug('Bin data', [\unreal4u\MQTT\DebugTools::convertToBinaryRepresentation($rawMQTTHeaders)]);
314
315			// Handy to have: the first byte
316			$firstByte = ord($fullMessage{0});
317			// TopicName size is always on the second position after the size of the remaining length field (1 to 4 bytes)
318			$topicSize = ord($fullMessage{$this->sizeOfRemainingLengthField + 2});
319			// With the first byte, we can determine the QoS level of the incoming message
320			$qosLevel = $this->determineIncomingQoSLevel($firstByte);
321
322			$messageStartPosition = $this->sizeOfRemainingLengthField + 3;
323			// If we have a QoS level present, we must retrieve the packet identifier as well
324			if ($qosLevel->getQoSLevel() > 0) {
325			$this->logger->debug('QoS level above 0, shifting message start position and getting packet identifier');
326			// [2 (fixed header) + 2 (topic size) + $topicSize] marks the beginning of the 2 packet identifier bytes
327			$this->setPacketIdentifier(new PacketIdentifier(Utilities::convertBinaryStringToNumber(
328			$fullMessage{$this->sizeOfRemainingLengthField + 3 + $topicSize} .
329			$fullMessage{$this->sizeOfRemainingLengthField + 4 + $topicSize}
330			)));
331			$this->logger->debug('Determined packet identifier', ['PI' => $this->getPacketIdentifier()]);
332			$messageStartPosition += 2;
333			}
334
335			// At this point $rawMQTTHeaders will be always 1 byte long, initialize a Message object with dummy data for now
336			$this->message = new Message(
337			// Save to assume a constant here: first 2 bytes will always be fixed header, next 2 bytes are topic size
338			substr($fullMessage, $messageStartPosition + $topicSize),
339			new TopicName(substr($fullMessage, $this->sizeOfRemainingLengthField + 3, $topicSize))
340			);
341			$this->analyzeFirstByte($firstByte, $qosLevel);
342
343			$this->logger->debug('Determined headers', [
344			'topicSize' => $topicSize,
345			'QoSLevel' => $this->message->getQoSLevel(),
346			'isDuplicate' => $this->isRedelivery,
347			'isRetained' => $this->message->isRetained(),
348			#'packetIdentifier' => $this->packetIdentifier->getPacketIdentifierValue(), // This is not always set!
349			]);
350
351			return $this;
352			}
353
354			/**
355			* @inheritdoc
356			* @throws InvalidRequest
357			* @throws InvalidQoSLevel
358			* @throws NotConnected
359			* @throws NoConnectionParametersDefined
360			*/
361			public function performSpecialActions(ClientInterface $client, WritableContentInterface $originalRequest): bool
362			{
363			$qosLevel = $this->message->getQoSLevel();
364			if ($qosLevel === 0) {
365			$this->logger->debug('No response needed', ['qosLevel', $qosLevel]);
366			} else {
367			if ($qosLevel === 1) {
368			$this->logger->debug('Responding with PubAck', ['qosLevel' => $qosLevel]);
369			$client->processObject($this->composePubAckAnswer());
370			} elseif ($qosLevel === 2) {
371			$this->logger->debug('Responding with PubRec', ['qosLevel' => $qosLevel]);
372			$client->processObject($this->composePubRecAnswer());
373			}
374			}
375
376			return true;
377			}
378
379			/**
380			* Composes a PubRec answer with the same packetIdentifier as what we received
381			*
382			* @return PubRec
383			* @throws InvalidRequest
384			*/
385			private function composePubRecAnswer(): PubRec
386			{
387			$this->checkForValidPacketIdentifier();
388			$pubRec = new PubRec($this->logger);
389			$pubRec->setPacketIdentifier($this->packetIdentifier);
390			return $pubRec;
391			}
392
393			/**
394			* Composes a PubAck answer with the same packetIdentifier as what we received
395			*
396			* @return PubAck
397			* @throws InvalidRequest
398			*/
399			private function composePubAckAnswer(): PubAck
400			{
401			$this->checkForValidPacketIdentifier();
402			$pubAck = new PubAck($this->logger);
403			$pubAck->setPacketIdentifier($this->packetIdentifier);
404			return $pubAck;
405			}
406
407			/**
408			* Will check whether the current object has a packet identifier set. If not, we are in serious problems!
409			*
410			* @return Publish
411			* @throws InvalidRequest
412			*/
413			private function checkForValidPacketIdentifier(): self
414			{
415			if ($this->packetIdentifier === null) {
416			$this->logger->critical('No valid packet identifier found at a stage where there MUST be one set');
417			throw new InvalidRequest('You are trying to send a request without a valid packet identifier');
418			}
419
420			return $this;
421			}
422
423			/**
424			* PUBLISH packet is the exception to the rule: it is not started on base of a packet that gets sent by us
425			*/
426			public function getOriginControlPacket(): int
427			{
428			return 0;
429			}
430			}
431

unreal4u / mqtt

Push — master ( d71bca...578a67 )

Publish::getMessage() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like