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