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.buffer.ByteBufAllocator;
21 import io.netty.buffer.Unpooled;
22 import io.netty.channel.ChannelHandler;
23 import io.netty.channel.ChannelHandlerContext;
24 import io.netty.handler.codec.EncoderException;
25 import io.netty.handler.codec.MessageToMessageEncoder;
26 import io.netty.util.internal.EmptyArrays;
27
28 import java.util.List;
29
30 import static io.netty.buffer.ByteBufUtil.*;
31 import static io.netty.handler.codec.mqtt.MqttCodecUtil.getMqttVersion;
32 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidClientId;
33 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidPublishTopicName;
34 import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidUserName;
35 import static io.netty.handler.codec.mqtt.MqttCodecUtil.setMqttVersion;
36 import static io.netty.handler.codec.mqtt.MqttConstant.DEFAULT_MAX_CLIENT_ID_LENGTH;
37 import static io.netty.handler.codec.mqtt.MqttProperties.ASSIGNED_CLIENT_IDENTIFIER;
38 import static io.netty.handler.codec.mqtt.MqttProperties.AUTHENTICATION_DATA;
39 import static io.netty.handler.codec.mqtt.MqttProperties.AUTHENTICATION_METHOD;
40 import static io.netty.handler.codec.mqtt.MqttProperties.CONTENT_TYPE;
41 import static io.netty.handler.codec.mqtt.MqttProperties.CORRELATION_DATA;
42 import static io.netty.handler.codec.mqtt.MqttProperties.MAXIMUM_PACKET_SIZE;
43 import static io.netty.handler.codec.mqtt.MqttProperties.MAXIMUM_QOS;
44 import static io.netty.handler.codec.mqtt.MqttProperties.PAYLOAD_FORMAT_INDICATOR;
45 import static io.netty.handler.codec.mqtt.MqttProperties.PUBLICATION_EXPIRY_INTERVAL;
46 import static io.netty.handler.codec.mqtt.MqttProperties.REASON_STRING;
47 import static io.netty.handler.codec.mqtt.MqttProperties.RECEIVE_MAXIMUM;
48 import static io.netty.handler.codec.mqtt.MqttProperties.REQUEST_PROBLEM_INFORMATION;
49 import static io.netty.handler.codec.mqtt.MqttProperties.REQUEST_RESPONSE_INFORMATION;
50 import static io.netty.handler.codec.mqtt.MqttProperties.RESPONSE_INFORMATION;
51 import static io.netty.handler.codec.mqtt.MqttProperties.RESPONSE_TOPIC;
52 import static io.netty.handler.codec.mqtt.MqttProperties.RETAIN_AVAILABLE;
53 import static io.netty.handler.codec.mqtt.MqttProperties.SERVER_KEEP_ALIVE;
54 import static io.netty.handler.codec.mqtt.MqttProperties.SERVER_REFERENCE;
55 import static io.netty.handler.codec.mqtt.MqttProperties.SESSION_EXPIRY_INTERVAL;
56 import static io.netty.handler.codec.mqtt.MqttProperties.SHARED_SUBSCRIPTION_AVAILABLE;
57 import static io.netty.handler.codec.mqtt.MqttProperties.SUBSCRIPTION_IDENTIFIER;
58 import static io.netty.handler.codec.mqtt.MqttProperties.SUBSCRIPTION_IDENTIFIER_AVAILABLE;
59 import static io.netty.handler.codec.mqtt.MqttProperties.TOPIC_ALIAS;
60 import static io.netty.handler.codec.mqtt.MqttProperties.TOPIC_ALIAS_MAXIMUM;
61 import static io.netty.handler.codec.mqtt.MqttProperties.USER_PROPERTY;
62 import static io.netty.handler.codec.mqtt.MqttProperties.WILDCARD_SUBSCRIPTION_AVAILABLE;
63 import static io.netty.handler.codec.mqtt.MqttProperties.WILL_DELAY_INTERVAL;
64
65
66
67
68
69
70
71 @ChannelHandler.Sharable
72 public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
73
74 public static final MqttEncoder INSTANCE = new MqttEncoder();
75
76
77 private static final int MAX_VARIABLE_LENGTH_INT = 268_435_455;
78
79 private MqttEncoder() {
80 super(MqttMessage.class);
81 }
82
83 @Override
84 protected void encode(ChannelHandlerContext ctx, MqttMessage msg, List<Object> out) throws Exception {
85 out.add(doEncode(ctx, msg));
86 }
87
88
89
90
91
92
93
94
95 static ByteBuf doEncode(ChannelHandlerContext ctx,
96 MqttMessage message) {
97
98 switch (message.fixedHeader().messageType()) {
99 case CONNECT:
100 return encodeConnectMessage(ctx, (MqttConnectMessage) message);
101
102 case CONNACK:
103 return encodeConnAckMessage(ctx, (MqttConnAckMessage) message);
104
105 case PUBLISH:
106 return encodePublishMessage(ctx, (MqttPublishMessage) message);
107
108 case SUBSCRIBE:
109 return encodeSubscribeMessage(ctx, (MqttSubscribeMessage) message);
110
111 case UNSUBSCRIBE:
112 return encodeUnsubscribeMessage(ctx, (MqttUnsubscribeMessage) message);
113
114 case SUBACK:
115 return encodeSubAckMessage(ctx, (MqttSubAckMessage) message);
116
117 case UNSUBACK:
118 if (message instanceof MqttUnsubAckMessage) {
119 return encodeUnsubAckMessage(ctx, (MqttUnsubAckMessage) message);
120 }
121 return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(ctx.alloc(), message);
122
123 case PUBACK:
124 case PUBREC:
125 case PUBREL:
126 case PUBCOMP:
127 return encodePubReplyMessage(ctx, message);
128
129 case DISCONNECT:
130 case AUTH:
131 return encodeReasonCodePlusPropertiesMessage(ctx, message);
132
133 case PINGREQ:
134 case PINGRESP:
135 return encodeMessageWithOnlySingleByteFixedHeader(ctx.alloc(), message);
136
137 default:
138 throw new IllegalArgumentException(
139 "Unknown message type: " + message.fixedHeader().messageType().value());
140 }
141 }
142
143 private static ByteBuf encodeConnectMessage(
144 ChannelHandlerContext ctx,
145 MqttConnectMessage message) {
146 int payloadBufferSize = 0;
147
148 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
149 MqttConnectVariableHeader variableHeader = message.variableHeader();
150 MqttConnectPayload payload = message.payload();
151 MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(variableHeader.name(),
152 (byte) variableHeader.version());
153 setMqttVersion(ctx, mqttVersion);
154
155
156 if ((mqttVersion == MqttVersion.MQTT_3_1 || mqttVersion == MqttVersion.MQTT_3_1_1) &&
157 !variableHeader.hasUserName() && variableHeader.hasPassword()) {
158 throw new EncoderException("Without a username, the password MUST be not set");
159 }
160
161
162 String clientIdentifier = payload.clientIdentifier();
163 if (!isValidClientId(mqttVersion, DEFAULT_MAX_CLIENT_ID_LENGTH, clientIdentifier, false)) {
164 throw new MqttIdentifierRejectedException("invalid clientIdentifier");
165 }
166 int clientIdentifierBytes = utf8Bytes(clientIdentifier);
167 payloadBufferSize += 2 + clientIdentifierBytes;
168
169
170 String willTopic = payload.willTopic();
171 int willTopicBytes = nullableUtf8Bytes(willTopic);
172 byte[] willMessage = payload.willMessageInBytes();
173 byte[] willMessageBytes = willMessage != null ? willMessage : EmptyArrays.EMPTY_BYTES;
174 if (variableHeader.isWillFlag()) {
175 if (!isValidPublishTopicName(willTopic)) {
176 throw new MqttIdentifierRejectedException("invalid willTopic");
177 }
178 payloadBufferSize += 2 + willTopicBytes;
179 payloadBufferSize += 2 + willMessageBytes.length;
180 }
181
182 String userName = payload.userName();
183 int userNameBytes = nullableUtf8Bytes(userName);
184 if (variableHeader.hasUserName()) {
185 if (!isValidUserName(userName)) {
186 throw new MqttIdentifierRejectedException("invalid userName");
187 }
188 payloadBufferSize += 2 + userNameBytes;
189 }
190
191 byte[] password = payload.passwordInBytes();
192 byte[] passwordBytes = password != null ? password : EmptyArrays.EMPTY_BYTES;
193 if (variableHeader.hasPassword()) {
194 payloadBufferSize += 2 + passwordBytes.length;
195 }
196
197
198 byte[] protocolNameBytes = mqttVersion.protocolNameBytes();
199 ByteBuf propertiesBuf = encodePropertiesIfNeeded(
200 mqttVersion,
201 ctx.alloc(),
202 message.variableHeader().properties());
203 try {
204 final ByteBuf willPropertiesBuf;
205 if (variableHeader.isWillFlag()) {
206 willPropertiesBuf = encodePropertiesIfNeeded(mqttVersion, ctx.alloc(), payload.willProperties());
207 payloadBufferSize += willPropertiesBuf.readableBytes();
208 } else {
209 willPropertiesBuf = Unpooled.EMPTY_BUFFER;
210 }
211 try {
212 int variableHeaderBufferSize = 2 + protocolNameBytes.length + 4 + propertiesBuf.readableBytes();
213
214 int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
215 int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
216 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
217 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
218 writeVariableLengthInt(buf, variablePartSize);
219
220 buf.writeShort(protocolNameBytes.length);
221 buf.writeBytes(protocolNameBytes);
222
223 buf.writeByte(variableHeader.version());
224 buf.writeByte(getConnVariableHeaderFlag(variableHeader));
225 buf.writeShort(variableHeader.keepAliveTimeSeconds());
226 buf.writeBytes(propertiesBuf);
227
228
229 writeExactUTF8String(buf, clientIdentifier, clientIdentifierBytes);
230 if (variableHeader.isWillFlag()) {
231 buf.writeBytes(willPropertiesBuf);
232 writeExactUTF8String(buf, willTopic, willTopicBytes);
233 buf.writeShort(willMessageBytes.length);
234 buf.writeBytes(willMessageBytes, 0, willMessageBytes.length);
235 }
236 if (variableHeader.hasUserName()) {
237 writeExactUTF8String(buf, userName, userNameBytes);
238 }
239 if (variableHeader.hasPassword()) {
240 buf.writeShort(passwordBytes.length);
241 buf.writeBytes(passwordBytes, 0, passwordBytes.length);
242 }
243 return buf;
244 } finally {
245 willPropertiesBuf.release();
246 }
247 } finally {
248 propertiesBuf.release();
249 }
250 }
251
252 private static int getConnVariableHeaderFlag(MqttConnectVariableHeader variableHeader) {
253 int flagByte = 0;
254 if (variableHeader.hasUserName()) {
255 flagByte |= 0x80;
256 }
257 if (variableHeader.hasPassword()) {
258 flagByte |= 0x40;
259 }
260 if (variableHeader.isWillRetain()) {
261 flagByte |= 0x20;
262 }
263 flagByte |= (variableHeader.willQos() & 0x03) << 3;
264 if (variableHeader.isWillFlag()) {
265 flagByte |= 0x04;
266 }
267 if (variableHeader.isCleanSession()) {
268 flagByte |= 0x02;
269 }
270 return flagByte;
271 }
272
273 private static ByteBuf encodeConnAckMessage(
274 ChannelHandlerContext ctx,
275 MqttConnAckMessage message) {
276 final MqttVersion mqttVersion = getMqttVersion(ctx);
277 ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
278 ctx.alloc(),
279 message.variableHeader().properties());
280
281 try {
282 final int variablePartSize = 2 + propertiesBuf.readableBytes();
283 final int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
284 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
285 buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
286 writeVariableLengthInt(buf, variablePartSize);
287 buf.writeByte(message.variableHeader().isSessionPresent() ? 0x01 : 0x00);
288 buf.writeByte(message.variableHeader().connectReturnCode().byteValue());
289 buf.writeBytes(propertiesBuf);
290 return buf;
291 } finally {
292 propertiesBuf.release();
293 }
294 }
295
296 private static ByteBuf encodeSubscribeMessage(
297 ChannelHandlerContext ctx,
298 MqttSubscribeMessage message) {
299 MqttVersion mqttVersion = getMqttVersion(ctx);
300 ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
301 ctx.alloc(),
302 message.idAndPropertiesVariableHeader().properties());
303
304 try {
305 final int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
306 int payloadBufferSize = 0;
307
308 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
309 MqttMessageIdVariableHeader variableHeader = message.variableHeader();
310 MqttSubscribePayload payload = message.payload();
311
312 for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
313 String topicName = topic.topicName();
314 int topicNameBytes = utf8Bytes(topicName);
315 payloadBufferSize += 2 + topicNameBytes;
316 payloadBufferSize += 1;
317 }
318
319 int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
320 int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
321
322 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
323 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
324 writeVariableLengthInt(buf, variablePartSize);
325
326
327 int messageId = variableHeader.messageId();
328 buf.writeShort(messageId);
329 buf.writeBytes(propertiesBuf);
330
331
332 for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
333 writeEagerUTF8String(buf, topic.topicName());
334 if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.MQTT_3_1) {
335 buf.writeByte(topic.qualityOfService().value());
336 } else {
337 final MqttSubscriptionOption option = topic.option();
338
339 int optionEncoded = option.retainHandling().value() << 4;
340 if (option.isRetainAsPublished()) {
341 optionEncoded |= 0x08;
342 }
343 if (option.isNoLocal()) {
344 optionEncoded |= 0x04;
345 }
346 optionEncoded |= option.qos().value();
347
348 buf.writeByte(optionEncoded);
349 }
350 }
351
352 return buf;
353 } finally {
354 propertiesBuf.release();
355 }
356 }
357
358 private static ByteBuf encodeUnsubscribeMessage(
359 ChannelHandlerContext ctx,
360 MqttUnsubscribeMessage message) {
361 MqttVersion mqttVersion = getMqttVersion(ctx);
362 ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
363 ctx.alloc(),
364 message.idAndPropertiesVariableHeader().properties());
365
366 try {
367 final int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
368 int payloadBufferSize = 0;
369
370 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
371 MqttMessageIdVariableHeader variableHeader = message.variableHeader();
372 MqttUnsubscribePayload payload = message.payload();
373
374 for (String topicName : payload.topics()) {
375 int topicNameBytes = utf8Bytes(topicName);
376 payloadBufferSize += 2 + topicNameBytes;
377 }
378
379 int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
380 int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
381
382 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
383 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
384 writeVariableLengthInt(buf, variablePartSize);
385
386
387 int messageId = variableHeader.messageId();
388 buf.writeShort(messageId);
389 buf.writeBytes(propertiesBuf);
390
391
392 for (String topicName : payload.topics()) {
393 writeEagerUTF8String(buf, topicName);
394 }
395
396 return buf;
397 } finally {
398 propertiesBuf.release();
399 }
400 }
401
402 private static ByteBuf encodeSubAckMessage(
403 ChannelHandlerContext ctx,
404 MqttSubAckMessage message) {
405 MqttVersion mqttVersion = getMqttVersion(ctx);
406 ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
407 ctx.alloc(),
408 message.idAndPropertiesVariableHeader().properties());
409 try {
410 int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
411 int payloadBufferSize = message.payload().grantedQoSLevels().size();
412 int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
413 int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
414 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
415 buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
416 writeVariableLengthInt(buf, variablePartSize);
417 buf.writeShort(message.variableHeader().messageId());
418 buf.writeBytes(propertiesBuf);
419 for (int code: message.payload().reasonCodes()) {
420 buf.writeByte(code);
421 }
422
423 return buf;
424 } finally {
425 propertiesBuf.release();
426 }
427 }
428
429 private static ByteBuf encodeUnsubAckMessage(
430 ChannelHandlerContext ctx,
431 MqttUnsubAckMessage message) {
432 if (message.variableHeader() instanceof MqttMessageIdAndPropertiesVariableHeader) {
433 MqttVersion mqttVersion = getMqttVersion(ctx);
434
435
436
437 final boolean writeReasonCodes = mqttVersion == MqttVersion.MQTT_5;
438 ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
439 ctx.alloc(),
440 message.idAndPropertiesVariableHeader().properties());
441 try {
442 int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
443 MqttUnsubAckPayload payload = message.payload();
444 int payloadBufferSize = writeReasonCodes && payload != null
445 ? payload.unsubscribeReasonCodes().size() : 0;
446 int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
447 int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
448 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
449 buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
450 writeVariableLengthInt(buf, variablePartSize);
451 buf.writeShort(message.variableHeader().messageId());
452 buf.writeBytes(propertiesBuf);
453
454 if (writeReasonCodes && payload != null) {
455 for (Short reasonCode : payload.unsubscribeReasonCodes()) {
456 buf.writeByte(reasonCode);
457 }
458 }
459
460 return buf;
461 } finally {
462 propertiesBuf.release();
463 }
464 } else {
465 return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(ctx.alloc(), message);
466 }
467 }
468
469 private static ByteBuf encodePublishMessage(
470 ChannelHandlerContext ctx,
471 MqttPublishMessage message) {
472 MqttVersion mqttVersion = getMqttVersion(ctx);
473 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
474 MqttPublishVariableHeader variableHeader = message.variableHeader();
475 ByteBuf payload = message.payload().duplicate();
476
477 String topicName = variableHeader.topicName();
478 if (!isValidPublishTopicName(topicName)) {
479 throw new MqttIdentifierRejectedException("invalid topicName");
480 }
481 int topicNameBytes = utf8Bytes(topicName);
482
483 ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
484 ctx.alloc(),
485 message.variableHeader().properties());
486
487 try {
488 boolean qosLevelGreaterZero = mqttFixedHeader.qosLevel().value() > 0;
489 int variableHeaderBufferSize = 2 + topicNameBytes +
490 (qosLevelGreaterZero ? 2 : 0) + propertiesBuf.readableBytes();
491 int payloadBufferSize = payload.readableBytes();
492 int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
493 int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
494
495 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
496 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
497 writeVariableLengthInt(buf, variablePartSize);
498 writeExactUTF8String(buf, topicName, topicNameBytes);
499 if (qosLevelGreaterZero) {
500 buf.writeShort(variableHeader.packetId());
501 }
502 buf.writeBytes(propertiesBuf);
503 buf.writeBytes(payload);
504
505 return buf;
506 } finally {
507 propertiesBuf.release();
508 }
509 }
510
511 private static ByteBuf encodePubReplyMessage(ChannelHandlerContext ctx,
512 MqttMessage message) {
513 if (message.variableHeader() instanceof MqttPubReplyMessageVariableHeader) {
514 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
515 MqttPubReplyMessageVariableHeader variableHeader =
516 (MqttPubReplyMessageVariableHeader) message.variableHeader();
517 int msgId = variableHeader.messageId();
518
519 final ByteBuf propertiesBuf;
520 final boolean includeReasonCode;
521 final int variableHeaderBufferSize;
522 final MqttVersion mqttVersion = getMqttVersion(ctx);
523 if (mqttVersion == MqttVersion.MQTT_5 &&
524 (variableHeader.reasonCode() != MqttPubReplyMessageVariableHeader.REASON_CODE_OK ||
525 !variableHeader.properties().isEmpty())) {
526 propertiesBuf = encodeProperties(ctx.alloc(), variableHeader.properties());
527 includeReasonCode = true;
528 variableHeaderBufferSize = 3 + propertiesBuf.readableBytes();
529 } else {
530 propertiesBuf = Unpooled.EMPTY_BUFFER;
531 includeReasonCode = false;
532 variableHeaderBufferSize = 2;
533 }
534
535 try {
536 final int fixedHeaderBufferSize = 1 + getVariableLengthInt(variableHeaderBufferSize);
537 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variableHeaderBufferSize);
538 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
539 writeVariableLengthInt(buf, variableHeaderBufferSize);
540 buf.writeShort(msgId);
541 if (includeReasonCode) {
542 buf.writeByte(variableHeader.reasonCode());
543 }
544 buf.writeBytes(propertiesBuf);
545
546 return buf;
547 } finally {
548 propertiesBuf.release();
549 }
550 } else {
551 return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(ctx.alloc(), message);
552 }
553 }
554
555 private static ByteBuf encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(
556 ByteBufAllocator byteBufAllocator,
557 MqttMessage message) {
558 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
559 MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) message.variableHeader();
560 int msgId = variableHeader.messageId();
561
562 int variableHeaderBufferSize = 2;
563 int fixedHeaderBufferSize = 1 + getVariableLengthInt(variableHeaderBufferSize);
564 ByteBuf buf = byteBufAllocator.buffer(fixedHeaderBufferSize + variableHeaderBufferSize);
565 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
566 writeVariableLengthInt(buf, variableHeaderBufferSize);
567 buf.writeShort(msgId);
568
569 return buf;
570 }
571
572 private static ByteBuf encodeReasonCodePlusPropertiesMessage(
573 ChannelHandlerContext ctx,
574 MqttMessage message) {
575 if (message.variableHeader() instanceof MqttReasonCodeAndPropertiesVariableHeader) {
576 MqttVersion mqttVersion = getMqttVersion(ctx);
577 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
578 MqttReasonCodeAndPropertiesVariableHeader variableHeader =
579 (MqttReasonCodeAndPropertiesVariableHeader) message.variableHeader();
580
581 final ByteBuf propertiesBuf;
582 final boolean includeReasonCode;
583 final int variableHeaderBufferSize;
584 if (mqttVersion == MqttVersion.MQTT_5 &&
585 (variableHeader.reasonCode() != MqttReasonCodeAndPropertiesVariableHeader.REASON_CODE_OK ||
586 !variableHeader.properties().isEmpty())) {
587 propertiesBuf = encodeProperties(ctx.alloc(), variableHeader.properties());
588 includeReasonCode = true;
589 variableHeaderBufferSize = 1 + propertiesBuf.readableBytes();
590 } else {
591 propertiesBuf = Unpooled.EMPTY_BUFFER;
592 includeReasonCode = false;
593 variableHeaderBufferSize = 0;
594 }
595
596 try {
597 final int fixedHeaderBufferSize = 1 + getVariableLengthInt(variableHeaderBufferSize);
598 ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variableHeaderBufferSize);
599 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
600 writeVariableLengthInt(buf, variableHeaderBufferSize);
601 if (includeReasonCode) {
602 buf.writeByte(variableHeader.reasonCode());
603 }
604 buf.writeBytes(propertiesBuf);
605
606 return buf;
607 } finally {
608 propertiesBuf.release();
609 }
610 } else {
611 return encodeMessageWithOnlySingleByteFixedHeader(ctx.alloc(), message);
612 }
613 }
614
615 private static ByteBuf encodeMessageWithOnlySingleByteFixedHeader(
616 ByteBufAllocator byteBufAllocator,
617 MqttMessage message) {
618 MqttFixedHeader mqttFixedHeader = message.fixedHeader();
619 ByteBuf buf = byteBufAllocator.buffer(2);
620 buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
621 buf.writeByte(0);
622
623 return buf;
624 }
625
626 private static ByteBuf encodePropertiesIfNeeded(MqttVersion mqttVersion,
627 ByteBufAllocator byteBufAllocator,
628 MqttProperties mqttProperties) {
629 if (mqttVersion == MqttVersion.MQTT_5) {
630 return encodeProperties(byteBufAllocator, mqttProperties);
631 }
632 return Unpooled.EMPTY_BUFFER;
633 }
634
635 private static ByteBuf encodeProperties(ByteBufAllocator byteBufAllocator,
636 MqttProperties mqttProperties) {
637 ByteBuf propertiesHeaderBuf = byteBufAllocator.buffer();
638
639 try {
640 ByteBuf propertiesBuf = byteBufAllocator.buffer();
641 try {
642 for (MqttProperties.MqttProperty property : mqttProperties.listAll()) {
643 int propertyId = property.propertyId;
644 switch (propertyId) {
645 case PAYLOAD_FORMAT_INDICATOR:
646 case REQUEST_PROBLEM_INFORMATION:
647 case REQUEST_RESPONSE_INFORMATION:
648 case MAXIMUM_QOS:
649 case RETAIN_AVAILABLE:
650 case WILDCARD_SUBSCRIPTION_AVAILABLE:
651 case SUBSCRIPTION_IDENTIFIER_AVAILABLE:
652 case SHARED_SUBSCRIPTION_AVAILABLE:
653 writeVariableLengthInt(propertiesBuf, propertyId);
654 final byte bytePropValue = ((MqttProperties.IntegerProperty) property).value.byteValue();
655 propertiesBuf.writeByte(bytePropValue);
656 break;
657 case SERVER_KEEP_ALIVE:
658 case RECEIVE_MAXIMUM:
659 case TOPIC_ALIAS_MAXIMUM:
660 case TOPIC_ALIAS:
661 writeVariableLengthInt(propertiesBuf, propertyId);
662 final short twoBytesInPropValue =
663 ((MqttProperties.IntegerProperty) property).value.shortValue();
664 propertiesBuf.writeShort(twoBytesInPropValue);
665 break;
666 case PUBLICATION_EXPIRY_INTERVAL:
667 case SESSION_EXPIRY_INTERVAL:
668 case WILL_DELAY_INTERVAL:
669 case MAXIMUM_PACKET_SIZE:
670 writeVariableLengthInt(propertiesBuf, propertyId);
671 final int fourBytesIntPropValue = ((MqttProperties.IntegerProperty) property).value;
672 propertiesBuf.writeInt(fourBytesIntPropValue);
673 break;
674 case SUBSCRIPTION_IDENTIFIER:
675 writeVariableLengthInt(propertiesBuf, propertyId);
676 final int vbi = ((MqttProperties.IntegerProperty) property).value;
677 writeVariableLengthInt(propertiesBuf, vbi);
678 break;
679 case CONTENT_TYPE:
680 case RESPONSE_TOPIC:
681 case ASSIGNED_CLIENT_IDENTIFIER:
682 case AUTHENTICATION_METHOD:
683 case RESPONSE_INFORMATION:
684 case SERVER_REFERENCE:
685 case REASON_STRING:
686 writeVariableLengthInt(propertiesBuf, propertyId);
687 writeEagerUTF8String(propertiesBuf, ((MqttProperties.StringProperty) property).value);
688 break;
689 case USER_PROPERTY:
690 final List<MqttProperties.StringPair> pairs =
691 ((MqttProperties.UserProperties) property).value;
692 for (MqttProperties.StringPair pair : pairs) {
693 writeVariableLengthInt(propertiesBuf, propertyId);
694 writeEagerUTF8String(propertiesBuf, pair.key);
695 writeEagerUTF8String(propertiesBuf, pair.value);
696 }
697 break;
698 case CORRELATION_DATA:
699 case AUTHENTICATION_DATA:
700 writeVariableLengthInt(propertiesBuf, propertyId);
701 final byte[] binaryPropValue = ((MqttProperties.BinaryProperty) property).value;
702 propertiesBuf.writeShort(binaryPropValue.length);
703 propertiesBuf.writeBytes(binaryPropValue, 0, binaryPropValue.length);
704 break;
705 default:
706
707 throw new EncoderException("Unknown property type: " + propertyId);
708 }
709 }
710 writeVariableLengthInt(propertiesHeaderBuf, propertiesBuf.readableBytes());
711 propertiesHeaderBuf.writeBytes(propertiesBuf);
712
713 return propertiesHeaderBuf;
714 } finally {
715 propertiesBuf.release();
716 }
717 } catch (RuntimeException e) {
718 propertiesHeaderBuf.release();
719 throw e;
720 }
721 }
722
723 private static int getFixedHeaderByte1(MqttFixedHeader header) {
724 int ret = 0;
725 ret |= header.messageType().value() << 4;
726 if (header.isDup()) {
727 ret |= 0x08;
728 }
729 ret |= header.qosLevel().value() << 1;
730 if (header.isRetain()) {
731 ret |= 0x01;
732 }
733 return ret;
734 }
735
736 private static void writeVariableLengthInt(ByteBuf buf, int num) {
737 checkVariableLengthInt(num);
738 do {
739 int digit = num & 0x7F;
740 num >>>= 7;
741 if (num > 0) {
742 digit |= 0x80;
743 }
744 buf.writeByte(digit);
745 } while (num > 0);
746 }
747
748 private static int nullableUtf8Bytes(String s) {
749 return s == null? 0 : utf8Bytes(s);
750 }
751
752 private static int nullableMaxUtf8Bytes(String s) {
753 return s == null? 0 : utf8MaxBytes(s);
754 }
755
756 private static void writeExactUTF8String(ByteBuf buf, String s, int utf8Length) {
757 buf.ensureWritable(utf8Length + 2);
758 buf.writeShort(utf8Length);
759 if (utf8Length > 0) {
760 final int writtenUtf8Length = reserveAndWriteUtf8(buf, s, utf8Length);
761 assert writtenUtf8Length == utf8Length;
762 }
763 }
764
765 private static void writeEagerUTF8String(ByteBuf buf, String s) {
766 final int maxUtf8Length = nullableMaxUtf8Bytes(s);
767 buf.ensureWritable(maxUtf8Length + 2);
768 final int writerIndex = buf.writerIndex();
769 final int startUtf8String = writerIndex + 2;
770 buf.writerIndex(startUtf8String);
771 final int utf8Length = s != null? reserveAndWriteUtf8(buf, s, maxUtf8Length) : 0;
772 buf.setShort(writerIndex, utf8Length);
773 }
774
775 private static void checkVariableLengthInt(int num) {
776
777 if (num < 0 || num > MAX_VARIABLE_LENGTH_INT) {
778 throw new EncoderException("Variable Byte Integer out of range: " + num +
779 " (max: " + MAX_VARIABLE_LENGTH_INT + ')');
780 }
781 }
782
783 private static int getVariableLengthInt(int num) {
784 checkVariableLengthInt(num);
785 if (num < 128) {
786 return 1;
787 }
788 if (num < 16_384) {
789 return 2;
790 }
791 if (num < 2_097_152) {
792 return 3;
793 }
794 return 4;
795 }
796
797 }