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.channel.ChannelHandlerContext;
20 import io.netty.handler.codec.DecoderException;
21 import io.netty.util.Attribute;
22 import io.netty.util.AttributeKey;
23
24 import static io.netty.handler.codec.mqtt.MqttConstant.MIN_CLIENT_ID_LENGTH;
25
26 final class MqttCodecUtil {
27
28 private static final char[] TOPIC_WILDCARDS = {'#', '+'};
29
30 static final AttributeKey<MqttVersion> MQTT_VERSION_KEY = AttributeKey.valueOf("NETTY_CODEC_MQTT_VERSION");
31
32 static MqttVersion getMqttVersion(ChannelHandlerContext ctx) {
33 Attribute<MqttVersion> attr = ctx.channel().attr(MQTT_VERSION_KEY);
34 MqttVersion version = attr.get();
35 if (version == null) {
36 return MqttVersion.MQTT_3_1_1;
37 }
38 return version;
39 }
40
41 static void setMqttVersion(ChannelHandlerContext ctx, MqttVersion version) {
42 Attribute<MqttVersion> attr = ctx.channel().attr(MQTT_VERSION_KEY);
43 attr.set(version);
44 }
45
46 static boolean isValidPublishTopicName(String topicName) {
47 if (topicName == null) {
48 return false;
49 }
50
51 for (int i = 0; i < topicName.length(); i++) {
52 char c = topicName.charAt(i);
53 if (c == '#' || c == '+' || c == '\0') {
54 return false;
55 }
56 }
57 return true;
58 }
59
60 static boolean isValidMessageId(int messageId) {
61 return messageId != 0;
62 }
63
64 static boolean isValidUserName(String userName) {
65 return userName == null || userName.indexOf('\0') == -1;
66 }
67
68
69
70
71
72
73
74
75
76
77 static boolean isValidClientId(MqttVersion mqttVersion, int maxClientIdLength, String clientId,
78 boolean acceptNulBytes) {
79 if (clientId == null || (!acceptNulBytes && clientId.indexOf('\0') != -1)) {
80 return false;
81 }
82 if (mqttVersion == MqttVersion.MQTT_3_1) {
83 return clientId.length() >= MIN_CLIENT_ID_LENGTH && clientId.length() <= maxClientIdLength;
84 }
85 if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.MQTT_5) {
86
87
88 return true;
89 }
90 throw new IllegalArgumentException(mqttVersion + " is unknown mqtt version");
91 }
92
93 static MqttFixedHeader validateFixedHeader(ChannelHandlerContext ctx, MqttFixedHeader mqttFixedHeader) {
94 switch (mqttFixedHeader.messageType()) {
95 case PUBREL:
96 case SUBSCRIBE:
97 case UNSUBSCRIBE:
98 if (mqttFixedHeader.qosLevel() != MqttQoS.AT_LEAST_ONCE) {
99 throw new DecoderException(mqttFixedHeader.messageType().name() + " message must have QoS 1");
100 }
101 return mqttFixedHeader;
102 case AUTH:
103 if (MqttCodecUtil.getMqttVersion(ctx) != MqttVersion.MQTT_5) {
104 throw new DecoderException("AUTH message requires at least MQTT 5");
105 }
106 return mqttFixedHeader;
107 default:
108 return mqttFixedHeader;
109 }
110 }
111
112 static MqttFixedHeader resetUnusedFields(MqttFixedHeader mqttFixedHeader) {
113 switch (mqttFixedHeader.messageType()) {
114 case CONNECT:
115 case CONNACK:
116 case PUBACK:
117 case PUBREC:
118 case PUBCOMP:
119 case SUBACK:
120 case UNSUBACK:
121 case PINGREQ:
122 case PINGRESP:
123 case DISCONNECT:
124 if (mqttFixedHeader.isDup() ||
125 mqttFixedHeader.qosLevel() != MqttQoS.AT_MOST_ONCE ||
126 mqttFixedHeader.isRetain()) {
127 return new MqttFixedHeader(
128 mqttFixedHeader.messageType(),
129 false,
130 MqttQoS.AT_MOST_ONCE,
131 false,
132 mqttFixedHeader.remainingLength());
133 }
134 return mqttFixedHeader;
135 case PUBREL:
136 case SUBSCRIBE:
137 case UNSUBSCRIBE:
138 if (mqttFixedHeader.isRetain()) {
139 return new MqttFixedHeader(
140 mqttFixedHeader.messageType(),
141 mqttFixedHeader.isDup(),
142 mqttFixedHeader.qosLevel(),
143 false,
144 mqttFixedHeader.remainingLength());
145 }
146 return mqttFixedHeader;
147 default:
148 return mqttFixedHeader;
149 }
150 }
151
152 private MqttCodecUtil() { }
153 }