1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package io.netty.handler.codec.mqtt;
18
19 import io.netty.buffer.ByteBuf;
20 import io.netty.channel.ChannelHandlerContext;
21 import io.netty.handler.codec.DecoderException;
22 import io.netty.handler.codec.ReplayingDecoder;
23 import io.netty.handler.codec.TooLongFrameException;
24 import io.netty.handler.codec.mqtt.MqttDecoder.DecoderState;
25 import io.netty.handler.codec.mqtt.MqttProperties.IntegerProperty;
26 import io.netty.util.CharsetUtil;
27 import io.netty.util.Signal;
28 import io.netty.util.internal.ObjectUtil;
29
30 import java.nio.ByteBuffer;
31 import java.nio.charset.CharacterCodingException;
32 import java.nio.charset.CharsetDecoder;
33 import java.nio.charset.CodingErrorAction;
34 import java.util.ArrayList;
35 import java.util.List;
36
37 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidClientId;
38 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidMessageId;
39 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidPublishTopicName;
40 import static io.netty.handler.codec.mqtt.MqttCodecUtil.resetUnusedFields;
41 import static io.netty.handler.codec.mqtt.MqttCodecUtil.validateFixedHeader;
42 import static io.netty.handler.codec.mqtt.MqttConstant.DEFAULT_MAX_BYTES_IN_MESSAGE;
43 import static io.netty.handler.codec.mqtt.MqttConstant.DEFAULT_MAX_CLIENT_ID_LENGTH;
44 import static io.netty.handler.codec.mqtt.MqttSubscriptionOption.RetainedHandlingPolicy;
45
46
47
48
49
50
51
52
53
54 public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
55
56
57
58
59
60
61 enum DecoderState {
62 READ_FIXED_HEADER,
63 READ_VARIABLE_HEADER,
64 READ_PAYLOAD,
65 BAD_MESSAGE,
66 }
67
68 private MqttFixedHeader mqttFixedHeader;
69 private Object variableHeader;
70 private int bytesRemainingInVariablePart;
71
72 private final int maxBytesInMessage;
73 private final int maxClientIdLength;
74 private final boolean strictUtf8Validation;
75
76
77
78 private CharsetDecoder utf8Decoder;
79
80 public MqttDecoder() {
81 this(DEFAULT_MAX_BYTES_IN_MESSAGE, DEFAULT_MAX_CLIENT_ID_LENGTH, true);
82 }
83
84 public MqttDecoder(int maxBytesInMessage) {
85 this(maxBytesInMessage, DEFAULT_MAX_CLIENT_ID_LENGTH, true);
86 }
87
88 public MqttDecoder(int maxBytesInMessage, int maxClientIdLength) {
89 this(maxBytesInMessage, maxClientIdLength, true);
90 }
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105 public MqttDecoder(int maxBytesInMessage, int maxClientIdLength, boolean strictUtf8Validation) {
106 super(DecoderState.READ_FIXED_HEADER);
107 this.maxBytesInMessage = ObjectUtil.checkPositive(maxBytesInMessage, "maxBytesInMessage");
108 this.maxClientIdLength = ObjectUtil.checkPositive(maxClientIdLength, "maxClientIdLength");
109 this.strictUtf8Validation = strictUtf8Validation;
110 }
111
112 @Override
113 protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
114 switch (state()) {
115 case READ_FIXED_HEADER: try {
116 mqttFixedHeader = decodeFixedHeader(ctx, buffer);
117 bytesRemainingInVariablePart = mqttFixedHeader.remainingLength();
118 checkpoint(DecoderState.READ_VARIABLE_HEADER);
119
120 } catch (Exception cause) {
121 out.add(invalidMessage(cause));
122 return;
123 }
124
125 case READ_VARIABLE_HEADER: try {
126 int bytesRemainingBeforeVariableHeader = bytesRemainingInVariablePart;
127 boolean bailOut = false;
128 try {
129 variableHeader = decodeVariableHeader(ctx, buffer, mqttFixedHeader);
130 } catch (Signal signal) {
131 if (bytesRemainingBeforeVariableHeader > maxBytesInMessage) {
132
133
134
135 bailOut = true;
136 } else {
137
138 throw signal;
139 }
140 }
141 if (bailOut || bytesRemainingBeforeVariableHeader > maxBytesInMessage) {
142 buffer.skipBytes(actualReadableBytes());
143 throw new TooLongFrameException("message length exceeds " + maxBytesInMessage + ": "
144 + bytesRemainingBeforeVariableHeader);
145 }
146 checkpoint(DecoderState.READ_PAYLOAD);
147
148 } catch (Exception cause) {
149 out.add(invalidMessage(cause));
150 return;
151 }
152
153 case READ_PAYLOAD: try {
154 final Object decodedPayload =
155 decodePayload(
156 ctx,
157 buffer,
158 mqttFixedHeader.messageType(),
159 maxClientIdLength,
160 variableHeader);
161 checkpoint(DecoderState.READ_FIXED_HEADER);
162 MqttMessage message = MqttMessageFactory.newMessage(
163 mqttFixedHeader, variableHeader, decodedPayload);
164 mqttFixedHeader = null;
165 variableHeader = null;
166 out.add(message);
167 break;
168 } catch (Exception cause) {
169 out.add(invalidMessage(cause));
170 return;
171 }
172
173 case BAD_MESSAGE:
174
175 buffer.skipBytes(actualReadableBytes());
176 break;
177
178 default:
179
180 throw new Error();
181 }
182 }
183
184 private MqttMessage invalidMessage(Throwable cause) {
185 checkpoint(DecoderState.BAD_MESSAGE);
186 return MqttMessageFactory.newInvalidMessage(mqttFixedHeader, variableHeader, cause);
187 }
188
189
190
191
192
193
194
195
196
197
198
199 private static MqttFixedHeader decodeFixedHeader(ChannelHandlerContext ctx, ByteBuf buffer) {
200 short b1 = buffer.readUnsignedByte();
201
202 MqttMessageType messageType = MqttMessageType.valueOf(b1 >> 4);
203 boolean dupFlag = (b1 & 0x08) == 0x08;
204 int qosLevel = (b1 & 0x06) >> 1;
205 boolean retain = (b1 & 0x01) != 0;
206
207 switch (messageType) {
208 case PUBLISH:
209 if (qosLevel == 3) {
210 throw new DecoderException("Illegal QOS Level in fixed header of PUBLISH message ("
211 + qosLevel + ')');
212 }
213 break;
214
215 case PUBREL:
216 case SUBSCRIBE:
217 case UNSUBSCRIBE:
218 if (dupFlag) {
219 throw new DecoderException("Illegal BIT 3 in fixed header of " + messageType
220 + " message, must be 0, found 1");
221 }
222 if (qosLevel != 1) {
223 throw new DecoderException("Illegal QOS Level in fixed header of " + messageType
224 + " message, must be 1, found " + qosLevel);
225 }
226 if (retain) {
227 throw new DecoderException("Illegal BIT 0 in fixed header of " + messageType
228 + " message, must be 0, found 1");
229 }
230 break;
231
232 case AUTH:
233 case CONNACK:
234 case CONNECT:
235 case DISCONNECT:
236 case PINGREQ:
237 case PINGRESP:
238 case PUBACK:
239 case PUBCOMP:
240 case PUBREC:
241 case SUBACK:
242 case UNSUBACK:
243 if (dupFlag) {
244 throw new DecoderException("Illegal BIT 3 in fixed header of " + messageType
245 + " message, must be 0, found 1");
246 }
247 if (qosLevel != 0) {
248 throw new DecoderException("Illegal BIT 2 or 1 in fixed header of " + messageType
249 + " message, must be 0, found " + qosLevel);
250 }
251 if (retain) {
252 throw new DecoderException("Illegal BIT 0 in fixed header of " + messageType
253 + " message, must be 0, found 1");
254 }
255 break;
256 default:
257 throw new DecoderException("Unknown message type, do not know how to validate fixed header");
258 }
259
260 int remainingLength = 0;
261 int multiplier = 1;
262 short digit;
263 int loops = 0;
264 do {
265 digit = buffer.readUnsignedByte();
266 remainingLength += (digit & 127) * multiplier;
267 multiplier *= 128;
268 loops++;
269 } while ((digit & 128) != 0 && loops < 4);
270
271
272 if (loops == 4 && (digit & 128) != 0) {
273 throw new DecoderException("remaining length exceeds 4 digits (" + messageType + ')');
274 }
275 MqttFixedHeader decodedFixedHeader =
276 new MqttFixedHeader(messageType, dupFlag, MqttQoS.valueOf(qosLevel), retain, remainingLength);
277 return validateFixedHeader(ctx, resetUnusedFields(decodedFixedHeader));
278 }
279
280
281
282
283
284
285
286 private Object decodeVariableHeader(ChannelHandlerContext ctx, ByteBuf buffer, MqttFixedHeader mqttFixedHeader) {
287 switch (mqttFixedHeader.messageType()) {
288 case CONNECT:
289 return decodeConnectionVariableHeader(ctx, buffer);
290
291 case CONNACK:
292 return decodeConnAckVariableHeader(ctx, buffer);
293
294 case UNSUBSCRIBE:
295 case SUBSCRIBE:
296 case SUBACK:
297 case UNSUBACK:
298 return decodeMessageIdAndPropertiesVariableHeader(ctx, buffer);
299
300 case PUBACK:
301 case PUBREC:
302 case PUBCOMP:
303 case PUBREL:
304 return decodePubReplyMessage(buffer);
305
306 case PUBLISH:
307 return decodePublishVariableHeader(ctx, buffer, mqttFixedHeader);
308
309 case DISCONNECT:
310 case AUTH:
311 return decodeReasonCodeAndPropertiesVariableHeader(buffer);
312
313 case PINGREQ:
314 case PINGRESP:
315
316 return null;
317 default:
318
319 throw new DecoderException("Unknown message type: " + mqttFixedHeader.messageType());
320 }
321 }
322
323 private MqttConnectVariableHeader decodeConnectionVariableHeader(
324 ChannelHandlerContext ctx,
325 ByteBuf buffer) {
326 final Result<String> protoString = decodeString(buffer);
327 int numberOfBytesConsumed = protoString.numberOfBytesConsumed;
328
329 final byte protocolLevel = buffer.readByte();
330 numberOfBytesConsumed += 1;
331
332 MqttVersion version = MqttVersion.fromProtocolNameAndLevel(protoString.value, protocolLevel);
333 MqttCodecUtil.setMqttVersion(ctx, version);
334
335 final int b1 = buffer.readUnsignedByte();
336 numberOfBytesConsumed += 1;
337
338 final int keepAlive = decodeMsbLsb(buffer);
339 numberOfBytesConsumed += 2;
340
341 final boolean hasUserName = (b1 & 0x80) == 0x80;
342 final boolean hasPassword = (b1 & 0x40) == 0x40;
343 final boolean willRetain = (b1 & 0x20) == 0x20;
344 final int willQos = (b1 & 0x18) >> 3;
345 final boolean willFlag = (b1 & 0x04) == 0x04;
346 final boolean cleanSession = (b1 & 0x02) == 0x02;
347 if (version == MqttVersion.MQTT_3_1_1 || version == MqttVersion.MQTT_5) {
348 final boolean zeroReservedFlag = (b1 & 0x01) == 0x0;
349 if (!zeroReservedFlag) {
350
351
352
353 throw new DecoderException("non-zero reserved flag");
354 }
355 }
356
357 final MqttProperties properties;
358 if (version == MqttVersion.MQTT_5) {
359 final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
360 properties = propertiesResult.value;
361 numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
362 } else {
363 properties = MqttProperties.NO_PROPERTIES;
364 }
365
366 bytesRemainingInVariablePart -= numberOfBytesConsumed;
367 return new MqttConnectVariableHeader(
368 version.protocolName(),
369 version.protocolLevel(),
370 hasUserName,
371 hasPassword,
372 willRetain,
373 willQos,
374 willFlag,
375 cleanSession,
376 keepAlive,
377 properties);
378 }
379
380 private MqttConnAckVariableHeader decodeConnAckVariableHeader(
381 ChannelHandlerContext ctx,
382 ByteBuf buffer) {
383 final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
384 final boolean sessionPresent = (buffer.readUnsignedByte() & 0x01) == 0x01;
385 byte returnCode = buffer.readByte();
386
387 final MqttProperties properties;
388 if (mqttVersion == MqttVersion.MQTT_5) {
389 final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
390 properties = propertiesResult.value;
391 bytesRemainingInVariablePart -= 2 + propertiesResult.numberOfBytesConsumed;
392 } else {
393 properties = MqttProperties.NO_PROPERTIES;
394 bytesRemainingInVariablePart -= 2;
395 }
396
397 return new MqttConnAckVariableHeader(MqttConnectReturnCode.valueOf(returnCode), sessionPresent, properties);
398 }
399
400 private MqttMessageIdAndPropertiesVariableHeader decodeMessageIdAndPropertiesVariableHeader(
401 ChannelHandlerContext ctx,
402 ByteBuf buffer) {
403 final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
404 final int packetId = decodeMessageId(buffer);
405
406 if (mqttVersion == MqttVersion.MQTT_5) {
407 final Result<MqttProperties> properties = decodeProperties(buffer);
408 bytesRemainingInVariablePart -= 2 + properties.numberOfBytesConsumed;
409 return new MqttMessageIdAndPropertiesVariableHeader(packetId, properties.value);
410 } else {
411 bytesRemainingInVariablePart -= 2;
412 return new MqttMessageIdAndPropertiesVariableHeader(packetId,
413 MqttProperties.NO_PROPERTIES);
414 }
415 }
416
417 private MqttPubReplyMessageVariableHeader decodePubReplyMessage(ByteBuf buffer) {
418 final int packetId = decodeMessageId(buffer);
419
420 final int packetIdNumberOfBytesConsumed = 2;
421 if (bytesRemainingInVariablePart > 3) {
422 final byte reasonCode = buffer.readByte();
423 final Result<MqttProperties> properties = decodeProperties(buffer);
424 bytesRemainingInVariablePart -= packetIdNumberOfBytesConsumed + 1 + properties.numberOfBytesConsumed;
425 return new MqttPubReplyMessageVariableHeader(packetId,
426 reasonCode,
427 properties.value);
428 } else if (bytesRemainingInVariablePart > 2) {
429 final byte reasonCode = buffer.readByte();
430 bytesRemainingInVariablePart -= packetIdNumberOfBytesConsumed + 1;
431 return new MqttPubReplyMessageVariableHeader(packetId,
432 reasonCode,
433 MqttProperties.NO_PROPERTIES);
434 } else {
435 bytesRemainingInVariablePart -= packetIdNumberOfBytesConsumed;
436 return new MqttPubReplyMessageVariableHeader(packetId,
437 (byte) 0,
438 MqttProperties.NO_PROPERTIES);
439 }
440 }
441
442 private MqttReasonCodeAndPropertiesVariableHeader decodeReasonCodeAndPropertiesVariableHeader(
443 ByteBuf buffer) {
444 final byte reasonCode;
445 final MqttProperties properties;
446 if (bytesRemainingInVariablePart > 1) {
447 reasonCode = buffer.readByte();
448 final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
449 properties = propertiesResult.value;
450 bytesRemainingInVariablePart -= 1 + propertiesResult.numberOfBytesConsumed;
451 } else if (bytesRemainingInVariablePart > 0) {
452 reasonCode = buffer.readByte();
453 properties = MqttProperties.NO_PROPERTIES;
454 --bytesRemainingInVariablePart;
455 } else {
456 reasonCode = 0;
457 properties = MqttProperties.NO_PROPERTIES;
458 }
459
460 return new MqttReasonCodeAndPropertiesVariableHeader(reasonCode, properties);
461 }
462
463 private MqttPublishVariableHeader decodePublishVariableHeader(
464 ChannelHandlerContext ctx,
465 ByteBuf buffer,
466 MqttFixedHeader mqttFixedHeader) {
467 final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
468 final Result<String> decodedTopic = decodeString(buffer);
469 if (!isValidPublishTopicName(decodedTopic.value)) {
470 throw new DecoderException("invalid publish topic name: " + decodedTopic.value + " (contains wildcards)");
471 }
472 int numberOfBytesConsumed = decodedTopic.numberOfBytesConsumed;
473
474 int messageId = -1;
475 if (mqttFixedHeader.qosLevel().value() > 0) {
476 messageId = decodeMessageId(buffer);
477 numberOfBytesConsumed += 2;
478 }
479
480 final MqttProperties properties;
481 if (mqttVersion == MqttVersion.MQTT_5) {
482 final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
483 properties = propertiesResult.value;
484 numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
485 } else {
486 properties = MqttProperties.NO_PROPERTIES;
487 }
488
489 bytesRemainingInVariablePart -= numberOfBytesConsumed;
490 return new MqttPublishVariableHeader(decodedTopic.value, messageId, properties);
491 }
492
493
494
495
496 private static int decodeMessageId(ByteBuf buffer) {
497 final int messageId = decodeMsbLsb(buffer);
498 if (!isValidMessageId(messageId)) {
499 throw new DecoderException("invalid messageId: " + messageId);
500 }
501 return messageId;
502 }
503
504
505
506
507
508
509
510
511
512 private Object decodePayload(
513 ChannelHandlerContext ctx,
514 ByteBuf buffer,
515 MqttMessageType messageType,
516 int maxClientIdLength,
517 Object variableHeader) {
518 switch (messageType) {
519 case CONNECT:
520 return decodeConnectionPayload(buffer, maxClientIdLength, (MqttConnectVariableHeader) variableHeader);
521
522 case SUBSCRIBE:
523 return decodeSubscribePayload(buffer);
524
525 case SUBACK:
526 return decodeSubackPayload(buffer);
527
528 case UNSUBSCRIBE:
529 return decodeUnsubscribePayload(buffer);
530
531 case UNSUBACK:
532 return decodeUnsubAckPayload(ctx, buffer);
533
534 case PUBLISH:
535 return decodePublishPayload(buffer);
536
537 default:
538
539
540
541
542 validateNoBytesRemain(0);
543 return null;
544 }
545 }
546
547 private MqttConnectPayload decodeConnectionPayload(
548 ByteBuf buffer,
549 int maxClientIdLength,
550 MqttConnectVariableHeader mqttConnectVariableHeader) {
551 final Result<String> decodedClientId = decodeString(buffer);
552 final String decodedClientIdValue = decodedClientId.value;
553 final MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(mqttConnectVariableHeader.name(),
554 (byte) mqttConnectVariableHeader.version());
555 if (!isValidClientId(mqttVersion, maxClientIdLength, decodedClientIdValue)) {
556 throw new MqttIdentifierRejectedException("invalid clientIdentifier: " + decodedClientIdValue);
557 }
558 int numberOfBytesConsumed = decodedClientId.numberOfBytesConsumed;
559
560 Result<String> decodedWillTopic = null;
561 byte[] decodedWillMessage = null;
562
563 final MqttProperties willProperties;
564 if (mqttConnectVariableHeader.isWillFlag()) {
565 if (mqttVersion == MqttVersion.MQTT_5) {
566 final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
567 willProperties = propertiesResult.value;
568 numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
569 } else {
570 willProperties = MqttProperties.NO_PROPERTIES;
571 }
572 decodedWillTopic = decodeString(buffer, 0, 32767);
573 numberOfBytesConsumed += decodedWillTopic.numberOfBytesConsumed;
574 decodedWillMessage = decodeByteArray(buffer);
575 numberOfBytesConsumed += decodedWillMessage.length + 2;
576 } else {
577 willProperties = MqttProperties.NO_PROPERTIES;
578 }
579 Result<String> decodedUserName = null;
580 byte[] decodedPassword = null;
581 if (mqttConnectVariableHeader.hasUserName()) {
582 decodedUserName = decodeString(buffer);
583 numberOfBytesConsumed += decodedUserName.numberOfBytesConsumed;
584 }
585 if (mqttConnectVariableHeader.hasPassword()) {
586 decodedPassword = decodeByteArray(buffer);
587 numberOfBytesConsumed += decodedPassword.length + 2;
588 }
589
590 validateNoBytesRemain(numberOfBytesConsumed);
591 return new MqttConnectPayload(
592 decodedClientId.value,
593 willProperties,
594 decodedWillTopic != null ? decodedWillTopic.value : null,
595 decodedWillMessage,
596 decodedUserName != null ? decodedUserName.value : null,
597 decodedPassword);
598 }
599
600 private MqttSubscribePayload decodeSubscribePayload(
601 ByteBuf buffer) {
602 final List<MqttTopicSubscription> subscribeTopics = new ArrayList<MqttTopicSubscription>();
603 int numberOfBytesConsumed = 0;
604 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
605 final Result<String> decodedTopicName = decodeString(buffer);
606 numberOfBytesConsumed += decodedTopicName.numberOfBytesConsumed;
607
608 final short optionByte = buffer.readUnsignedByte();
609
610 MqttQoS qos = MqttQoS.valueOf(optionByte & 0x03);
611 boolean noLocal = ((optionByte & 0x04) >> 2) == 1;
612 boolean retainAsPublished = ((optionByte & 0x08) >> 3) == 1;
613 RetainedHandlingPolicy retainHandling = RetainedHandlingPolicy.valueOf((optionByte & 0x30) >> 4);
614
615 final MqttSubscriptionOption subscriptionOption = new MqttSubscriptionOption(qos,
616 noLocal,
617 retainAsPublished,
618 retainHandling);
619
620 numberOfBytesConsumed++;
621 subscribeTopics.add(new MqttTopicSubscription(decodedTopicName.value, subscriptionOption));
622 }
623 validateNoBytesRemain(numberOfBytesConsumed);
624 return new MqttSubscribePayload(subscribeTopics);
625 }
626
627 private MqttSubAckPayload decodeSubackPayload(
628 ByteBuf buffer) {
629 int bytesRemainingInVariablePart = this.bytesRemainingInVariablePart;
630 final List<Integer> grantedQos = new ArrayList<Integer>(bytesRemainingInVariablePart);
631 int numberOfBytesConsumed = 0;
632 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
633 int reasonCode = buffer.readUnsignedByte();
634 numberOfBytesConsumed++;
635 grantedQos.add(reasonCode);
636 }
637 validateNoBytesRemain(numberOfBytesConsumed);
638 return new MqttSubAckPayload(grantedQos);
639 }
640
641 private MqttUnsubAckPayload decodeUnsubAckPayload(
642 ChannelHandlerContext ctx,
643 ByteBuf buffer) {
644 int bytesRemainingInVariablePart = this.bytesRemainingInVariablePart;
645 final List<Short> reasonCodes = new ArrayList<Short>(bytesRemainingInVariablePart);
646 int numberOfBytesConsumed = 0;
647 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
648 short reasonCode = buffer.readUnsignedByte();
649 numberOfBytesConsumed++;
650 reasonCodes.add(reasonCode);
651 }
652 validateNoBytesRemain(numberOfBytesConsumed);
653 return new MqttUnsubAckPayload(reasonCodes);
654 }
655
656 private MqttUnsubscribePayload decodeUnsubscribePayload(
657 ByteBuf buffer) {
658 final List<String> unsubscribeTopics = new ArrayList<String>();
659 int numberOfBytesConsumed = 0;
660 while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
661 final Result<String> decodedTopicName = decodeString(buffer);
662 numberOfBytesConsumed += decodedTopicName.numberOfBytesConsumed;
663 unsubscribeTopics.add(decodedTopicName.value);
664 }
665 validateNoBytesRemain(numberOfBytesConsumed);
666 return new MqttUnsubscribePayload(unsubscribeTopics);
667 }
668
669 private ByteBuf decodePublishPayload(ByteBuf buffer) {
670 return buffer.readRetainedSlice(bytesRemainingInVariablePart);
671 }
672
673 private void validateNoBytesRemain(int numberOfBytesConsumed) {
674 bytesRemainingInVariablePart -= numberOfBytesConsumed;
675 if (bytesRemainingInVariablePart != 0) {
676 throw new DecoderException(
677 "non-zero remaining payload bytes: " +
678 bytesRemainingInVariablePart + " (" + mqttFixedHeader.messageType() + ')');
679 }
680 }
681
682 private Result<String> decodeString(ByteBuf buffer) {
683 return decodeString(buffer, 0, Integer.MAX_VALUE);
684 }
685
686 private Result<String> decodeString(ByteBuf buffer, int minBytes, int maxBytes) {
687 int size = decodeMsbLsb(buffer);
688 int numberOfBytesConsumed = 2;
689 if (size < minBytes || size > maxBytes) {
690 buffer.skipBytes(size);
691 numberOfBytesConsumed += size;
692 return new Result<String>(null, numberOfBytesConsumed);
693 }
694 final String s;
695 if (strictUtf8Validation) {
696 s = readStrictUtf8(buffer, size);
697 } else {
698 s = buffer.toString(buffer.readerIndex(), size, CharsetUtil.UTF_8);
699 buffer.skipBytes(size);
700 }
701 numberOfBytesConsumed += size;
702 return new Result<String>(s, numberOfBytesConsumed);
703 }
704
705
706
707
708
709
710 private String readStrictUtf8(ByteBuf buffer, int length) {
711 if (length == 0) {
712 return "";
713 }
714 final int readerIndex = buffer.readerIndex();
715 final ByteBuffer nioBuf;
716 if (buffer.nioBufferCount() == 1) {
717 nioBuf = buffer.nioBuffer(readerIndex, length);
718 } else {
719
720
721 byte[] tmp = new byte[length];
722 buffer.getBytes(readerIndex, tmp);
723 nioBuf = ByteBuffer.wrap(tmp);
724 }
725 if (utf8Decoder == null) {
726 utf8Decoder = CharsetUtil.UTF_8.newDecoder()
727 .onMalformedInput(CodingErrorAction.REPORT)
728 .onUnmappableCharacter(CodingErrorAction.REPORT);
729 }
730 utf8Decoder.reset();
731 final String s;
732 try {
733 s = utf8Decoder.decode(nioBuf).toString();
734 } catch (CharacterCodingException e) {
735 buffer.skipBytes(length);
736 throw new DecoderException("invalid UTF-8 string in MQTT packet", e);
737 }
738 buffer.skipBytes(length);
739
740
741 if (s.indexOf('\u0000') >= 0) {
742 throw new DecoderException("MQTT UTF-8 Encoded String must not contain U+0000");
743 }
744 return s;
745 }
746
747
748
749
750
751 private static byte[] decodeByteArray(ByteBuf buffer) {
752 int size = decodeMsbLsb(buffer);
753 byte[] bytes = new byte[size];
754 buffer.readBytes(bytes);
755 return bytes;
756 }
757
758
759 private static long packInts(int a, int b) {
760 return (((long) a) << 32) | (b & 0xFFFFFFFFL);
761 }
762
763 private static int unpackA(long ints) {
764 return (int) (ints >> 32);
765 }
766
767 private static int unpackB(long ints) {
768 return (int) ints;
769 }
770
771
772
773
774 private static int decodeMsbLsb(ByteBuf buffer) {
775 int min = 0;
776 int max = 65535;
777 short msbSize = buffer.readUnsignedByte();
778 short lsbSize = buffer.readUnsignedByte();
779 int result = msbSize << 8 | lsbSize;
780 if (result < min || result > max) {
781 result = -1;
782 }
783 return result;
784 }
785
786
787
788
789
790
791
792
793 private static long decodeVariableByteInteger(ByteBuf buffer) {
794 int remainingLength = 0;
795 int multiplier = 1;
796 short digit;
797 int loops = 0;
798 do {
799 digit = buffer.readUnsignedByte();
800 remainingLength += (digit & 127) * multiplier;
801 multiplier *= 128;
802 loops++;
803 } while ((digit & 128) != 0 && loops < 4);
804
805 if (loops == 4 && (digit & 128) != 0) {
806 throw new DecoderException("MQTT protocol limits Remaining Length to 4 bytes");
807 }
808 return packInts(remainingLength, loops);
809 }
810
811 private static final class Result<T> {
812
813 private final T value;
814 private final int numberOfBytesConsumed;
815
816 Result(T value, int numberOfBytesConsumed) {
817 this.value = value;
818 this.numberOfBytesConsumed = numberOfBytesConsumed;
819 }
820 }
821
822 private Result<MqttProperties> decodeProperties(ByteBuf buffer) {
823 final long propertiesLength = decodeVariableByteInteger(buffer);
824 int totalPropertiesLength = unpackA(propertiesLength);
825 int numberOfBytesConsumed = unpackB(propertiesLength);
826 if (totalPropertiesLength > 0) {
827
828
829
830
831
832
833 buffer.getByte(buffer.readerIndex() + totalPropertiesLength - 1);
834 }
835
836 MqttProperties decodedProperties = new MqttProperties();
837 while (numberOfBytesConsumed < totalPropertiesLength) {
838 long propertyId = decodeVariableByteInteger(buffer);
839 final int propertyIdValue = unpackA(propertyId);
840 numberOfBytesConsumed += unpackB(propertyId);
841 MqttProperties.MqttPropertyType propertyType = MqttProperties.MqttPropertyType.valueOf(propertyIdValue);
842 switch (propertyType) {
843 case PAYLOAD_FORMAT_INDICATOR:
844 case REQUEST_PROBLEM_INFORMATION:
845 case REQUEST_RESPONSE_INFORMATION:
846 case MAXIMUM_QOS:
847 case RETAIN_AVAILABLE:
848 case WILDCARD_SUBSCRIPTION_AVAILABLE:
849 case SUBSCRIPTION_IDENTIFIER_AVAILABLE:
850 case SHARED_SUBSCRIPTION_AVAILABLE:
851 final int b1 = buffer.readUnsignedByte();
852 numberOfBytesConsumed++;
853 decodedProperties.add(new IntegerProperty(propertyIdValue, b1));
854 break;
855 case SERVER_KEEP_ALIVE:
856 case RECEIVE_MAXIMUM:
857 case TOPIC_ALIAS_MAXIMUM:
858 case TOPIC_ALIAS:
859 final int int2BytesResult = decodeMsbLsb(buffer);
860 numberOfBytesConsumed += 2;
861 decodedProperties.add(new IntegerProperty(propertyIdValue, int2BytesResult));
862 break;
863 case PUBLICATION_EXPIRY_INTERVAL:
864 case SESSION_EXPIRY_INTERVAL:
865 case WILL_DELAY_INTERVAL:
866 case MAXIMUM_PACKET_SIZE:
867 final int maxPacketSize = buffer.readInt();
868 numberOfBytesConsumed += 4;
869 decodedProperties.add(new IntegerProperty(propertyIdValue, maxPacketSize));
870 break;
871 case SUBSCRIPTION_IDENTIFIER:
872 long vbIntegerResult = decodeVariableByteInteger(buffer);
873 numberOfBytesConsumed += unpackB(vbIntegerResult);
874 decodedProperties.add(new IntegerProperty(propertyIdValue, unpackA(vbIntegerResult)));
875 break;
876 case CONTENT_TYPE:
877 case RESPONSE_TOPIC:
878 case ASSIGNED_CLIENT_IDENTIFIER:
879 case AUTHENTICATION_METHOD:
880 case RESPONSE_INFORMATION:
881 case SERVER_REFERENCE:
882 case REASON_STRING:
883 final Result<String> stringResult = decodeString(buffer);
884 numberOfBytesConsumed += stringResult.numberOfBytesConsumed;
885 decodedProperties.add(new MqttProperties.StringProperty(propertyIdValue, stringResult.value));
886 break;
887 case USER_PROPERTY:
888 final Result<String> keyResult = decodeString(buffer);
889 final Result<String> valueResult = decodeString(buffer);
890 numberOfBytesConsumed += keyResult.numberOfBytesConsumed;
891 numberOfBytesConsumed += valueResult.numberOfBytesConsumed;
892 decodedProperties.add(new MqttProperties.UserProperty(keyResult.value, valueResult.value));
893 break;
894 case CORRELATION_DATA:
895 case AUTHENTICATION_DATA:
896 final byte[] binaryDataResult = decodeByteArray(buffer);
897 numberOfBytesConsumed += binaryDataResult.length + 2;
898 decodedProperties.add(new MqttProperties.BinaryProperty(propertyIdValue, binaryDataResult));
899 break;
900 default:
901
902 throw new DecoderException("Unknown property type: " + propertyType);
903 }
904 }
905
906 return new Result<MqttProperties>(decodedProperties, numberOfBytesConsumed);
907 }
908 }