升级netty版本为4.1.68.Final

This commit is contained in:
heavyrian2012
2022-09-18 22:05:22 +08:00
parent 279530ade2
commit 7ea1a2b146
44 changed files with 2710 additions and 448 deletions
@@ -112,7 +112,7 @@ abstract public class AdminAction extends Action {
}
protected void sendApiMessage(Response response, String fromUser, String clientId, String topic, byte[] message, ApiCallback callback, boolean noAdmin) {
ServerAPIHelper.sendRequest(fromUser, clientId, topic, message, new ServerAPIHelper.Callback() {
ServerAPIHelper.sendRequest(fromUser, clientId, topic, message, callback == null ? null : new ServerAPIHelper.Callback() {
@Override
public void onSuccess(byte[] result) {
if(callback != null) {
@@ -81,18 +81,6 @@ public class Request {
// request URI parameters
this.putParams(new QueryStringDecoder(uri));
if(nettyRequest.method() != HttpMethod.GET){
HttpPostRequestDecoder decoder = null;
try {
decoder = new HttpPostRequestDecoder(HTTP_DATA_FACTORY, nettyRequest);
this.putParams(decoder);
} finally {
if(null != decoder){
decoder.destroy();
decoder = null;
}
}
}
// IP
this.putIp(ctx);
@@ -72,7 +72,7 @@ public class InterceptConnectMessage extends InterceptAbstractMessage {
}
public byte[] getPassword() {
return msg.payload().password();
return msg.payload().passwordInBytes();
}
public String getWillTopic() {
@@ -308,7 +308,7 @@ public class ProtocolProcessor {
}
byte[] pwd = null;
if (msg.variableHeader().hasPassword()) {
pwd = msg.payload().password();
pwd = msg.payload().passwordInBytes();
MemorySessionStore.Session session = m_sessionsStore.getSession(clientId);
if (session == null) {
@@ -164,7 +164,7 @@ public class Qos1PublishHandler extends QosPublishHandler {
} else {
MemorySessionStore.Session session = m_sessionStore.getSession(clientID);
if (session != null && session.getUsername().equals(fromUser)) {
if (data.length > 7*1024 && session.getMqttVersion().protocolLevel() >= MqttVersion.Wildfire_1.protocolLevel()) {
if (data.length > 7*1024 && session.getMqttVersion().protocolLevel() >= MqttVersion.MQTT_5.protocolLevel()) {
ByteArrayOutputStream out = new ByteArrayOutputStream();
GZIPOutputStream gzip = null;
try {
+32 -9
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -16,13 +16,30 @@
package io.netty.handler.codec.mqtt;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.DecoderException;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
final class MqttCodecUtil {
private static final char[] TOPIC_WILDCARDS = {'#', '+'};
private static final int MIN_CLIENT_ID_LENGTH = 1;
private static final int MAX_CLIENT_ID_LENGTH = 23;
static final AttributeKey<MqttVersion> MQTT_VERSION_KEY = AttributeKey.valueOf("NETTY_CODEC_MQTT_VERSION");
static MqttVersion getMqttVersion(ChannelHandlerContext ctx) {
Attribute<MqttVersion> attr = ctx.channel().attr(MQTT_VERSION_KEY);
MqttVersion version = attr.get();
if (version == null) {
return MqttVersion.MQTT_3_1_1;
}
return version;
}
static void setMqttVersion(ChannelHandlerContext ctx, MqttVersion version) {
Attribute<MqttVersion> attr = ctx.channel().attr(MQTT_VERSION_KEY);
attr.set(version);
}
static boolean isValidPublishTopicName(String topicName) {
// publish topic name must not contain any wildcard
@@ -38,20 +55,20 @@ final class MqttCodecUtil {
return messageId != 0;
}
static boolean isValidClientId(MqttVersion mqttVersion, String clientId) {
static boolean isValidClientId(MqttVersion mqttVersion, int maxClientIdLength, String clientId) {
if (mqttVersion == MqttVersion.MQTT_3_1) {
return clientId != null && clientId.length() >= MIN_CLIENT_ID_LENGTH &&
clientId.length() <= MAX_CLIENT_ID_LENGTH;
return clientId != null && clientId.length() >= MqttConstant.MIN_CLIENT_ID_LENGTH &&
clientId.length() <= maxClientIdLength;
}
if (mqttVersion.protocolLevel() >= MqttVersion.MQTT_3_1_1.protocolLevel()) {
// In 3.1.3.1 Client Identifier of MQTT 3.1.1 specification, The Server MAY allow ClientId’s
// In 3.1.3.1 Client Identifier of MQTT 3.1.1 and 5.0 specifications, The Server MAY allow ClientId’s
// that contain more than 23 encoded bytes. And, The Server MAY allow zero-length ClientId.
return clientId != null;
}
throw new IllegalArgumentException(mqttVersion + " is unknown mqtt version");
}
static MqttFixedHeader validateFixedHeader(MqttFixedHeader mqttFixedHeader) {
static MqttFixedHeader validateFixedHeader(ChannelHandlerContext ctx, MqttFixedHeader mqttFixedHeader) {
switch (mqttFixedHeader.messageType()) {
case PUBREL:
case SUBSCRIBE:
@@ -59,6 +76,12 @@ final class MqttCodecUtil {
if (mqttFixedHeader.qosLevel() != MqttQoS.AT_LEAST_ONCE) {
throw new DecoderException(mqttFixedHeader.messageType().name() + " message must have QoS 1");
}
return mqttFixedHeader;
case AUTH:
if (MqttCodecUtil.getMqttVersion(ctx) != MqttVersion.MQTT_5) {
throw new DecoderException("AUTH message requires at least MQTT 5");
}
return mqttFixedHeader;
default:
return mqttFixedHeader;
}
@@ -75,7 +98,7 @@ final class MqttCodecUtil {
case UNSUBACK:
case PINGREQ:
case PINGRESP:
//case DISCONNECT:
case DISCONNECT:
if (mqttFixedHeader.isDup() ||
mqttFixedHeader.qosLevel() != MqttQoS.AT_MOST_ONCE ||
mqttFixedHeader.isRetain()) {
+3 -3
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -17,7 +17,7 @@
package io.netty.handler.codec.mqtt;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connack">MQTTV3.1/connack</a>
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connack">MQTTV3.1/connack</a>
*/
public final class MqttConnAckMessage extends MqttMessage {
@@ -28,7 +28,7 @@ public final class MqttConnAckMessage extends MqttMessage {
public MqttConnAckMessage(MqttFixedHeader mqttFixedHeader, MqttConnAckVariableHeader variableHeader, MqttConnectAckPayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
@Override
public MqttConnAckVariableHeader variableHeader() {
return (MqttConnAckVariableHeader) super.variableHeader();
+13 -1
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -27,9 +27,17 @@ public final class MqttConnAckVariableHeader {
private final boolean sessionPresent;
private final MqttProperties properties;
public MqttConnAckVariableHeader(MqttConnectReturnCode connectReturnCode, boolean sessionPresent) {
this(connectReturnCode, sessionPresent, MqttProperties.NO_PROPERTIES);
}
public MqttConnAckVariableHeader(MqttConnectReturnCode connectReturnCode, boolean sessionPresent,
MqttProperties properties) {
this.connectReturnCode = connectReturnCode;
this.sessionPresent = sessionPresent;
this.properties = MqttProperties.withEmptyDefaults(properties);
}
public MqttConnectReturnCode connectReturnCode() {
@@ -40,6 +48,10 @@ public final class MqttConnAckVariableHeader {
return sessionPresent;
}
public MqttProperties properties() {
return properties;
}
@Override
public String toString() {
return new StringBuilder(StringUtil.simpleClassName(this))
+2 -2
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -17,7 +17,7 @@
package io.netty.handler.codec.mqtt;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connect">MQTTV3.1/connect</a>
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connect">MQTTV3.1/connect</a>
*/
public final class MqttConnectMessage extends MqttMessage {
+65 -5
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -16,6 +16,9 @@
package io.netty.handler.codec.mqtt;
import java.util.Arrays;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.StringUtil;
/**
@@ -24,18 +27,55 @@ import io.netty.util.internal.StringUtil;
public final class MqttConnectPayload {
private final String clientIdentifier;
private final MqttProperties willProperties;
private final String willTopic;
private final String willMessage;
private final byte[] willMessage;
private final String userName;
private final byte[] password;
/**
* @deprecated use {@link MqttConnectPayload#MqttConnectPayload(String,
* MqttProperties, String, byte[], String, byte[])} instead
*/
@Deprecated
public MqttConnectPayload(
String clientIdentifier,
String willTopic,
String willMessage,
String userName,
String password) {
this(
clientIdentifier,
MqttProperties.NO_PROPERTIES,
willTopic,
willMessage.getBytes(CharsetUtil.UTF_8),
userName,
password.getBytes(CharsetUtil.UTF_8));
}
public MqttConnectPayload(
String clientIdentifier,
String willTopic,
byte[] willMessage,
String userName,
byte[] password) {
this(clientIdentifier,
MqttProperties.NO_PROPERTIES,
willTopic,
willMessage,
userName,
password);
}
public MqttConnectPayload(
String clientIdentifier,
MqttProperties willProperties,
String willTopic,
byte[] willMessage,
String userName,
byte[] password) {
this.clientIdentifier = clientIdentifier;
this.willProperties = MqttProperties.withEmptyDefaults(willProperties);
this.willTopic = willTopic;
this.willMessage = willMessage;
this.userName = userName;
@@ -46,11 +86,23 @@ public final class MqttConnectPayload {
return clientIdentifier;
}
public MqttProperties willProperties() {
return willProperties;
}
public String willTopic() {
return willTopic;
}
/**
* @deprecated use {@link MqttConnectPayload#willMessageInBytes()} instead
*/
@Deprecated
public String willMessage() {
return willMessage == null ? null : new String(willMessage, CharsetUtil.UTF_8);
}
public byte[] willMessageInBytes() {
return willMessage;
}
@@ -58,7 +110,15 @@ public final class MqttConnectPayload {
return userName;
}
public byte[] password() {
/**
* @deprecated use {@link MqttConnectPayload#passwordInBytes()} instead
*/
@Deprecated
public String password() {
return password == null ? null : new String(password, CharsetUtil.UTF_8);
}
public byte[] passwordInBytes() {
return password;
}
@@ -68,9 +128,9 @@ public final class MqttConnectPayload {
.append('[')
.append("clientIdentifier=").append(clientIdentifier)
.append(", willTopic=").append(willTopic)
.append(", willMessage=").append(willMessage)
.append(", willMessage=").append(Arrays.toString(willMessage))
.append(", userName=").append(userName)
.append(", password=").append(password)
.append(", password=").append(Arrays.toString(password))
.append(']')
.toString();
}
+42 -14
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -16,31 +16,52 @@
package io.netty.handler.codec.mqtt;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* Return Code of {@link MqttConnAckMessage}
*/
public enum MqttConnectReturnCode {
CONNECTION_ACCEPTED((byte) 0x00),
//MQTT 3 codes
CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION((byte) 0X01),
CONNECTION_REFUSED_IDENTIFIER_REJECTED((byte) 0x02),
CONNECTION_REFUSED_SERVER_UNAVAILABLE((byte) 0x03),
CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD((byte) 0x04),
CONNECTION_REFUSED_NOT_AUTHORIZED((byte) 0x05),
CONNECTION_REFUSED_UNEXPECT_NODE((byte) 0X06),
CONNECTION_REFUSED_SESSION_NOT_EXIST((byte) 0x07);
CONNECTION_REFUSED_SESSION_NOT_EXIST((byte) 0x07),
//MQTT 5 codes
CONNECTION_REFUSED_UNSPECIFIED_ERROR((byte) 0x80),
CONNECTION_REFUSED_MALFORMED_PACKET((byte) 0x81),
CONNECTION_REFUSED_PROTOCOL_ERROR((byte) 0x82),
CONNECTION_REFUSED_IMPLEMENTATION_SPECIFIC((byte) 0x83),
CONNECTION_REFUSED_UNSUPPORTED_PROTOCOL_VERSION((byte) 0x84),
CONNECTION_REFUSED_CLIENT_IDENTIFIER_NOT_VALID((byte) 0x85),
CONNECTION_REFUSED_BAD_USERNAME_OR_PASSWORD((byte) 0x86),
CONNECTION_REFUSED_NOT_AUTHORIZED_5((byte) 0x87),
CONNECTION_REFUSED_SERVER_UNAVAILABLE_5((byte) 0x88),
CONNECTION_REFUSED_SERVER_BUSY((byte) 0x89),
CONNECTION_REFUSED_BANNED((byte) 0x8A),
CONNECTION_REFUSED_BAD_AUTHENTICATION_METHOD((byte) 0x8C),
CONNECTION_REFUSED_TOPIC_NAME_INVALID((byte) 0x90),
CONNECTION_REFUSED_PACKET_TOO_LARGE((byte) 0x95),
CONNECTION_REFUSED_QUOTA_EXCEEDED((byte) 0x97),
CONNECTION_REFUSED_PAYLOAD_FORMAT_INVALID((byte) 0x99),
CONNECTION_REFUSED_RETAIN_NOT_SUPPORTED((byte) 0x9A),
CONNECTION_REFUSED_QOS_NOT_SUPPORTED((byte) 0x9B),
CONNECTION_REFUSED_USE_ANOTHER_SERVER((byte) 0x9C),
CONNECTION_REFUSED_SERVER_MOVED((byte) 0x9D),
CONNECTION_REFUSED_CONNECTION_RATE_EXCEEDED((byte) 0x9F);
private static final Map<Byte, MqttConnectReturnCode> VALUE_TO_CODE_MAP;
private static final MqttConnectReturnCode[] VALUES;
static {
final Map<Byte, MqttConnectReturnCode> valueMap = new HashMap<Byte, MqttConnectReturnCode>();
for (MqttConnectReturnCode code: values()) {
valueMap.put(code.byteValue, code);
MqttConnectReturnCode[] values = values();
VALUES = new MqttConnectReturnCode[160];
for (MqttConnectReturnCode code : values) {
final int unsignedByte = code.byteValue & 0xFF;
// Suppress a warning about out of bounds access since the enum contains only correct values
VALUES[unsignedByte] = code; // lgtm [java/index-out-of-bounds]
}
VALUE_TO_CODE_MAP = Collections.unmodifiableMap(valueMap);
}
private final byte byteValue;
@@ -54,9 +75,16 @@ public enum MqttConnectReturnCode {
}
public static MqttConnectReturnCode valueOf(byte b) {
if (VALUE_TO_CODE_MAP.containsKey(b)) {
return VALUE_TO_CODE_MAP.get(b);
final int unsignedByte = b & 0xFF;
MqttConnectReturnCode mqttConnectReturnCode = null;
try {
mqttConnectReturnCode = VALUES[unsignedByte];
} catch (ArrayIndexOutOfBoundsException ignored) {
// no op
}
throw new IllegalArgumentException("unknown connect return code: " + (b & 0xFF));
if (mqttConnectReturnCode == null) {
throw new IllegalArgumentException("unknown connect return code: " + unsignedByte);
}
return mqttConnectReturnCode;
}
}
+30 -1
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -32,6 +32,7 @@ public final class MqttConnectVariableHeader {
private final boolean isWillFlag;
private final boolean isCleanSession;
private final int keepAliveTimeSeconds;
private final MqttProperties properties;
public MqttConnectVariableHeader(
String name,
@@ -43,6 +44,29 @@ public final class MqttConnectVariableHeader {
boolean isWillFlag,
boolean isCleanSession,
int keepAliveTimeSeconds) {
this(name,
version,
hasUserName,
hasPassword,
isWillRetain,
willQos,
isWillFlag,
isCleanSession,
keepAliveTimeSeconds,
MqttProperties.NO_PROPERTIES);
}
public MqttConnectVariableHeader(
String name,
int version,
boolean hasUserName,
boolean hasPassword,
boolean isWillRetain,
int willQos,
boolean isWillFlag,
boolean isCleanSession,
int keepAliveTimeSeconds,
MqttProperties properties) {
this.name = name;
this.version = version;
this.hasUserName = hasUserName;
@@ -52,6 +76,7 @@ public final class MqttConnectVariableHeader {
this.isWillFlag = isWillFlag;
this.isCleanSession = isCleanSession;
this.keepAliveTimeSeconds = keepAliveTimeSeconds;
this.properties = MqttProperties.withEmptyDefaults(properties);
}
public String name() {
@@ -90,6 +115,10 @@ public final class MqttConnectVariableHeader {
return keepAliveTimeSeconds;
}
public MqttProperties properties() {
return properties;
}
@Override
public String toString() {
return new StringBuilder(StringUtil.simpleClassName(this))
@@ -0,0 +1,39 @@
/*
* Copyright 2021 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.handler.codec.mqtt;
public final class MqttConstant {
private MqttConstant() {
}
/**
* Default max bytes in message
*/
public static final int DEFAULT_MAX_BYTES_IN_MESSAGE = 262144; //256KB
/**
* min client id length
*/
public static final int MIN_CLIENT_ID_LENGTH = 1;
/**
* Default max client id length,In the mqtt3.1 protocol,
* the default maximum Client Identifier length is 23
*/
public static final int DEFAULT_MAX_CLIENT_ID_LENGTH = 23;
}
+436 -112
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -20,26 +20,22 @@ import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.DecoderException;
import io.netty.handler.codec.ReplayingDecoder;
import io.netty.handler.codec.mqtt.MqttDecoder.DecoderState;
import io.netty.handler.codec.TooLongFrameException;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.ObjectUtil;
import java.util.ArrayList;
import java.util.List;
import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidClientId;
import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidMessageId;
import static io.netty.handler.codec.mqtt.MqttCodecUtil.isValidPublishTopicName;
import static io.netty.handler.codec.mqtt.MqttCodecUtil.resetUnusedFields;
import static io.netty.handler.codec.mqtt.MqttCodecUtil.validateFixedHeader;
/**
* Decodes Mqtt messages from bytes, following
* <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html">
* the MQTT protocol specification v3.1</a>
* the MQTT protocol specification
* <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html">v3.1</a>
* or
* <a href="https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html">v5.0</a>, depending on the
* version specified in the CONNECT message that first goes through the channel.
*/
public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
private static final int DEFAULT_MAX_BYTES_IN_MESSAGE = 262144; //256K
public final class MqttDecoder extends ReplayingDecoder<MqttDecoder.DecoderState> {
/**
* States of the decoder.
@@ -58,31 +54,42 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
private int bytesRemainingInVariablePart;
private final int maxBytesInMessage;
private final int maxClientIdLength;
public MqttDecoder() {
this(DEFAULT_MAX_BYTES_IN_MESSAGE);
this(MqttConstant.DEFAULT_MAX_BYTES_IN_MESSAGE, MqttConstant.DEFAULT_MAX_CLIENT_ID_LENGTH);
}
public MqttDecoder(int maxBytesInMessage) {
this(maxBytesInMessage, MqttConstant.DEFAULT_MAX_CLIENT_ID_LENGTH);
}
public MqttDecoder(int maxBytesInMessage, int maxClientIdLength) {
super(DecoderState.READ_FIXED_HEADER);
this.maxBytesInMessage = maxBytesInMessage;
this.maxBytesInMessage = ObjectUtil.checkPositive(maxBytesInMessage, "maxBytesInMessage");
this.maxClientIdLength = ObjectUtil.checkPositive(maxClientIdLength, "maxClientIdLength");
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
switch (state()) {
case READ_FIXED_HEADER:
mqttFixedHeader = decodeFixedHeader(buffer);
case READ_FIXED_HEADER: try {
mqttFixedHeader = decodeFixedHeader(ctx, buffer);
bytesRemainingInVariablePart = mqttFixedHeader.remainingLength();
checkpoint(DecoderState.READ_VARIABLE_HEADER);
// fall through
} catch (Exception cause) {
out.add(invalidMessage(cause));
return;
}
case READ_VARIABLE_HEADER: try {
if (bytesRemainingInVariablePart > maxBytesInMessage) {
throw new DecoderException("too large message: " + bytesRemainingInVariablePart + " bytes");
}
final Result<?> decodedVariableHeader = decodeVariableHeader(buffer, mqttFixedHeader);
final Result<?> decodedVariableHeader = decodeVariableHeader(ctx, buffer, mqttFixedHeader);
variableHeader = decodedVariableHeader.value;
if (bytesRemainingInVariablePart > maxBytesInMessage) {
buffer.skipBytes(actualReadableBytes());
throw new TooLongFrameException("too large message: " + bytesRemainingInVariablePart + " bytes");
}
bytesRemainingInVariablePart -= decodedVariableHeader.numberOfBytesConsumed;
checkpoint(DecoderState.READ_PAYLOAD);
// fall through
@@ -94,9 +101,11 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
case READ_PAYLOAD: try {
final Result<?> decodedPayload =
decodePayload(
ctx,
buffer,
mqttFixedHeader.messageType(),
bytesRemainingInVariablePart,
maxClientIdLength,
variableHeader);
bytesRemainingInVariablePart -= decodedPayload.numberOfBytesConsumed;
if (bytesRemainingInVariablePart != 0) {
@@ -129,16 +138,20 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
private MqttMessage invalidMessage(Throwable cause) {
checkpoint(DecoderState.BAD_MESSAGE);
return MqttMessageFactory.newInvalidMessage(cause);
return MqttMessageFactory.newInvalidMessage(mqttFixedHeader, variableHeader, cause);
}
/**
* Decodes the fixed header. It's one byte for the flags and then variable bytes for the remaining length.
* Decodes the fixed header. It's one byte for the flags and then variable
* bytes for the remaining length.
*
* @see
* https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html#_Toc442180841
*
* @param buffer the buffer to decode from
* @return the fixed header
*/
private static MqttFixedHeader decodeFixedHeader(ByteBuf buffer) {
private static MqttFixedHeader decodeFixedHeader(ChannelHandlerContext ctx, ByteBuf buffer) {
short b1 = buffer.readUnsignedByte();
MqttMessageType messageType = MqttMessageType.valueOf(b1 >> 4);
@@ -146,6 +159,59 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
int qosLevel = (b1 & 0x06) >> 1;
boolean retain = (b1 & 0x01) != 0;
switch (messageType) {
case PUBLISH:
if (qosLevel == 3) {
throw new DecoderException("Illegal QOS Level in fixed header of PUBLISH message ("
+ qosLevel + ')');
}
break;
case PUBREL:
case SUBSCRIBE:
case UNSUBSCRIBE:
if (dupFlag) {
throw new DecoderException("Illegal BIT 3 in fixed header of " + messageType
+ " message, must be 0, found 1");
}
if (qosLevel != 1) {
throw new DecoderException("Illegal QOS Level in fixed header of " + messageType
+ " message, must be 1, found " + qosLevel);
}
if (retain) {
throw new DecoderException("Illegal BIT 0 in fixed header of " + messageType
+ " message, must be 0, found 1");
}
break;
case AUTH:
case CONNACK:
case CONNECT:
case DISCONNECT:
case PINGREQ:
case PINGRESP:
case PUBACK:
case PUBCOMP:
case PUBREC:
case SUBACK:
case UNSUBACK:
if (dupFlag) {
throw new DecoderException("Illegal BIT 3 in fixed header of " + messageType
+ " message, must be 0, found 1");
}
if (qosLevel != 0) {
throw new DecoderException("Illegal BIT 2 or 1 in fixed header of " + messageType
+ " message, must be 0, found " + qosLevel);
}
if (retain) {
throw new DecoderException("Illegal BIT 0 in fixed header of " + messageType
+ " message, must be 0, found 1");
}
break;
default:
throw new DecoderException("Unknown message type, do not know how to validate fixed header");
}
int remainingLength = 0;
int multiplier = 1;
short digit;
@@ -163,7 +229,7 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
}
MqttFixedHeader decodedFixedHeader =
new MqttFixedHeader(messageType, dupFlag, MqttQoS.valueOf(qosLevel), retain, remainingLength);
return validateFixedHeader(resetUnusedFields(decodedFixedHeader));
return MqttCodecUtil.validateFixedHeader(ctx, MqttCodecUtil.resetUnusedFields(decodedFixedHeader));
}
/**
@@ -172,50 +238,60 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
* @param mqttFixedHeader MqttFixedHeader of the same message
* @return the variable header
*/
private static Result<?> decodeVariableHeader(ByteBuf buffer, MqttFixedHeader mqttFixedHeader) {
private Result<?> decodeVariableHeader(ChannelHandlerContext ctx, ByteBuf buffer, MqttFixedHeader mqttFixedHeader) {
switch (mqttFixedHeader.messageType()) {
case CONNECT:
return decodeConnectionVariableHeader(buffer);
return decodeConnectionVariableHeader(ctx, buffer);
case CONNACK:
return decodeConnAckVariableHeader(buffer);
return decodeConnAckVariableHeader(ctx, buffer);
case SUBSCRIBE:
case UNSUBSCRIBE:
case SUBSCRIBE:
case SUBACK:
case UNSUBACK:
return decodeMessageIdAndPropertiesVariableHeader(ctx, buffer);
case PUBACK:
case PUBREC:
case PUBCOMP:
case PUBREL:
return decodeMessageIdVariableHeader(buffer);
return decodePubReplyMessage(buffer);
case PUBLISH:
return decodePublishVariableHeader(buffer, mqttFixedHeader);
return decodePublishVariableHeader(ctx, buffer, mqttFixedHeader);
case DISCONNECT:
case AUTH:
return decodeReasonCodeAndPropertiesVariableHeader(buffer);
case PINGREQ:
case PINGRESP:
case DISCONNECT:
// Empty variable header
return new Result<Object>(null, 0);
default:
//shouldn't reach here
throw new DecoderException("Unknown message type: " + mqttFixedHeader.messageType());
}
return new Result<Object>(null, 0); //should never reach here
}
private static Result<MqttConnectVariableHeader> decodeConnectionVariableHeader(ByteBuf buffer) {
private static Result<MqttConnectVariableHeader> decodeConnectionVariableHeader(
ChannelHandlerContext ctx,
ByteBuf buffer) {
final Result<String> protoString = decodeString(buffer);
int numberOfBytesConsumed = protoString.numberOfBytesConsumed;
final byte protocolLevel = buffer.readByte();
numberOfBytesConsumed += 1;
final MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(protoString.value, protocolLevel);
MqttVersion version = MqttVersion.fromProtocolNameAndLevel(protoString.value, protocolLevel);
MqttCodecUtil.setMqttVersion(ctx, version);
final int b1 = buffer.readUnsignedByte();
numberOfBytesConsumed += 1;
final Result<Integer> keepAlive = decodeMsbLsb(buffer);
numberOfBytesConsumed += keepAlive.numberOfBytesConsumed;
final int keepAlive = decodeMsbLsb(buffer);
numberOfBytesConsumed += 2;
final boolean hasUserName = (b1 & 0x80) == 0x80;
final boolean hasPassword = (b1 & 0x40) == 0x40;
@@ -223,69 +299,178 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
final int willQos = (b1 & 0x18) >> 3;
final boolean willFlag = (b1 & 0x04) == 0x04;
final boolean cleanSession = (b1 & 0x02) == 0x02;
if (mqttVersion.protocolLevel() >= MqttVersion.MQTT_3_1_1.protocolLevel()) {
if (version == MqttVersion.MQTT_3_1_1 || version == MqttVersion.MQTT_5) {
final boolean zeroReservedFlag = (b1 & 0x01) == 0x0;
if (!zeroReservedFlag) {
// MQTT v3.1.1: The Server MUST validate that the reserved flag in the CONNECT Control Packet is
// set to zero and disconnect the Client if it is not zero.
// See http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc385349230
// See https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc385349230
throw new DecoderException("non-zero reserved flag");
}
}
final MqttProperties properties;
if (version == MqttVersion.MQTT_5) {
final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
properties = propertiesResult.value;
numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
} else {
properties = MqttProperties.NO_PROPERTIES;
}
final MqttConnectVariableHeader mqttConnectVariableHeader = new MqttConnectVariableHeader(
mqttVersion.protocolName(),
mqttVersion.protocolLevel(),
version.protocolName(),
version.protocolLevel(),
hasUserName,
hasPassword,
willRetain,
willQos,
willFlag,
cleanSession,
keepAlive.value);
keepAlive,
properties);
return new Result<MqttConnectVariableHeader>(mqttConnectVariableHeader, numberOfBytesConsumed);
}
private static Result<MqttConnAckVariableHeader> decodeConnAckVariableHeader(ByteBuf buffer) {
private static Result<MqttConnAckVariableHeader> decodeConnAckVariableHeader(
ChannelHandlerContext ctx,
ByteBuf buffer) {
final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
final boolean sessionPresent = (buffer.readUnsignedByte() & 0x01) == 0x01;
byte returnCode = buffer.readByte();
final int numberOfBytesConsumed = 2;
int numberOfBytesConsumed = 2;
final MqttProperties properties;
if (mqttVersion == MqttVersion.MQTT_5) {
final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
properties = propertiesResult.value;
numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
} else {
properties = MqttProperties.NO_PROPERTIES;
}
final MqttConnAckVariableHeader mqttConnAckVariableHeader =
new MqttConnAckVariableHeader(MqttConnectReturnCode.valueOf(returnCode), sessionPresent);
new MqttConnAckVariableHeader(MqttConnectReturnCode.valueOf(returnCode), sessionPresent, properties);
return new Result<MqttConnAckVariableHeader>(mqttConnAckVariableHeader, numberOfBytesConsumed);
}
private static Result<MqttMessageIdVariableHeader> decodeMessageIdVariableHeader(ByteBuf buffer) {
final Result<Integer> messageId = decodeMessageId(buffer);
return new Result<MqttMessageIdVariableHeader>(
MqttMessageIdVariableHeader.from(messageId.value),
messageId.numberOfBytesConsumed);
private static Result<MqttMessageIdAndPropertiesVariableHeader> decodeMessageIdAndPropertiesVariableHeader(
ChannelHandlerContext ctx,
ByteBuf buffer) {
final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
final int packetId = decodeMessageId(buffer);
final MqttMessageIdAndPropertiesVariableHeader mqttVariableHeader;
final int mqtt5Consumed;
if (mqttVersion == MqttVersion.MQTT_5) {
final Result<MqttProperties> properties = decodeProperties(buffer);
mqttVariableHeader = new MqttMessageIdAndPropertiesVariableHeader(packetId, properties.value);
mqtt5Consumed = properties.numberOfBytesConsumed;
} else {
mqttVariableHeader = new MqttMessageIdAndPropertiesVariableHeader(packetId,
MqttProperties.NO_PROPERTIES);
mqtt5Consumed = 0;
}
return new Result<MqttMessageIdAndPropertiesVariableHeader>(mqttVariableHeader,
2 + mqtt5Consumed);
}
private static Result<MqttPublishVariableHeader> decodePublishVariableHeader(
private Result<MqttPubReplyMessageVariableHeader> decodePubReplyMessage(ByteBuf buffer) {
final int packetId = decodeMessageId(buffer);
final MqttPubReplyMessageVariableHeader mqttPubAckVariableHeader;
final int consumed;
final int packetIdNumberOfBytesConsumed = 2;
if (bytesRemainingInVariablePart > 3) {
final byte reasonCode = buffer.readByte();
final Result<MqttProperties> properties = decodeProperties(buffer);
mqttPubAckVariableHeader = new MqttPubReplyMessageVariableHeader(packetId,
reasonCode,
properties.value);
consumed = packetIdNumberOfBytesConsumed + 1 + properties.numberOfBytesConsumed;
} else if (bytesRemainingInVariablePart > 2) {
final byte reasonCode = buffer.readByte();
mqttPubAckVariableHeader = new MqttPubReplyMessageVariableHeader(packetId,
reasonCode,
MqttProperties.NO_PROPERTIES);
consumed = packetIdNumberOfBytesConsumed + 1;
} else {
mqttPubAckVariableHeader = new MqttPubReplyMessageVariableHeader(packetId,
(byte) 0,
MqttProperties.NO_PROPERTIES);
consumed = packetIdNumberOfBytesConsumed;
}
return new Result<MqttPubReplyMessageVariableHeader>(mqttPubAckVariableHeader, consumed);
}
private Result<MqttReasonCodeAndPropertiesVariableHeader> decodeReasonCodeAndPropertiesVariableHeader(
ByteBuf buffer) {
final byte reasonCode;
final MqttProperties properties;
final int consumed;
if (bytesRemainingInVariablePart > 1) {
reasonCode = buffer.readByte();
final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
properties = propertiesResult.value;
consumed = 1 + propertiesResult.numberOfBytesConsumed;
} else if (bytesRemainingInVariablePart > 0) {
reasonCode = buffer.readByte();
properties = MqttProperties.NO_PROPERTIES;
consumed = 1;
} else {
reasonCode = 0;
properties = MqttProperties.NO_PROPERTIES;
consumed = 0;
}
final MqttReasonCodeAndPropertiesVariableHeader mqttReasonAndPropsVariableHeader =
new MqttReasonCodeAndPropertiesVariableHeader(reasonCode, properties);
return new Result<MqttReasonCodeAndPropertiesVariableHeader>(
mqttReasonAndPropsVariableHeader,
consumed);
}
private Result<MqttPublishVariableHeader> decodePublishVariableHeader(
ChannelHandlerContext ctx,
ByteBuf buffer,
MqttFixedHeader mqttFixedHeader) {
final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
final Result<String> decodedTopic = decodeString(buffer);
if (!isValidPublishTopicName(decodedTopic.value)) {
if (!MqttCodecUtil.isValidPublishTopicName(decodedTopic.value)) {
throw new DecoderException("invalid publish topic name: " + decodedTopic.value + " (contains wildcards)");
}
int numberOfBytesConsumed = decodedTopic.numberOfBytesConsumed;
int messageId = -1;
if (mqttFixedHeader.qosLevel().value() > 0) {
final Result<Integer> decodedMessageId = decodeMessageId(buffer);
messageId = decodedMessageId.value;
numberOfBytesConsumed += decodedMessageId.numberOfBytesConsumed;
messageId = decodeMessageId(buffer);
numberOfBytesConsumed += 2;
}
final MqttProperties properties;
if (mqttVersion == MqttVersion.MQTT_5) {
final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
properties = propertiesResult.value;
numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
} else {
properties = MqttProperties.NO_PROPERTIES;
}
final MqttPublishVariableHeader mqttPublishVariableHeader =
new MqttPublishVariableHeader(decodedTopic.value, messageId);
new MqttPublishVariableHeader(decodedTopic.value, messageId, properties);
return new Result<MqttPublishVariableHeader>(mqttPublishVariableHeader, numberOfBytesConsumed);
}
private static Result<Integer> decodeMessageId(ByteBuf buffer) {
final Result<Integer> messageId = decodeMsbLsb(buffer);
if (!isValidMessageId(messageId.value)) {
throw new DecoderException("invalid messageId: " + messageId.value);
/**
* @return messageId with numberOfBytesConsumed is 2
*/
private static int decodeMessageId(ByteBuf buffer) {
final int messageId = decodeMsbLsb(buffer);
if (!MqttCodecUtil.isValidMessageId(messageId)) {
throw new DecoderException("invalid messageId: " + messageId);
}
return messageId;
}
@@ -300,13 +485,15 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
* @return the payload
*/
private static Result<?> decodePayload(
ChannelHandlerContext ctx,
ByteBuf buffer,
MqttMessageType messageType,
int bytesRemainingInVariablePart,
int maxClientIdLength,
Object variableHeader) {
switch (messageType) {
case CONNECT:
return decodeConnectionPayload(buffer, (MqttConnectVariableHeader) variableHeader);
return decodeConnectionPayload(buffer, maxClientIdLength, (MqttConnectVariableHeader) variableHeader);
case SUBSCRIBE:
return decodeSubscribePayload(buffer, bytesRemainingInVariablePart);
@@ -317,6 +504,9 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
case UNSUBSCRIBE:
return decodeUnsubscribePayload(buffer, bytesRemainingInVariablePart);
case UNSUBACK:
return decodeUnsubAckPayload(ctx, buffer, bytesRemainingInVariablePart);
case PUBLISH:
return decodePublishPayload(buffer, bytesRemainingInVariablePart);
@@ -328,42 +518,55 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
private static Result<MqttConnectPayload> decodeConnectionPayload(
ByteBuf buffer,
int maxClientIdLength,
MqttConnectVariableHeader mqttConnectVariableHeader) {
final Result<String> decodedClientId = decodeString(buffer);
final String decodedClientIdValue = decodedClientId.value;
final MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(mqttConnectVariableHeader.name(),
(byte) mqttConnectVariableHeader.version());
if (!isValidClientId(mqttVersion, decodedClientIdValue)) {
if (!MqttCodecUtil.isValidClientId(mqttVersion, maxClientIdLength, decodedClientIdValue)) {
throw new MqttIdentifierRejectedException("invalid clientIdentifier: " + decodedClientIdValue);
}
int numberOfBytesConsumed = decodedClientId.numberOfBytesConsumed;
Result<String> decodedWillTopic = null;
Result<String> decodedWillMessage = null;
byte[] decodedWillMessage = null;
final MqttProperties willProperties;
if (mqttConnectVariableHeader.isWillFlag()) {
if (mqttVersion == MqttVersion.MQTT_5) {
final Result<MqttProperties> propertiesResult = decodeProperties(buffer);
willProperties = propertiesResult.value;
numberOfBytesConsumed += propertiesResult.numberOfBytesConsumed;
} else {
willProperties = MqttProperties.NO_PROPERTIES;
}
decodedWillTopic = decodeString(buffer, 0, 32767);
numberOfBytesConsumed += decodedWillTopic.numberOfBytesConsumed;
decodedWillMessage = decodeAsciiString(buffer);
numberOfBytesConsumed += decodedWillMessage.numberOfBytesConsumed;
decodedWillMessage = decodeByteArray(buffer);
numberOfBytesConsumed += decodedWillMessage.length + 2;
} else {
willProperties = MqttProperties.NO_PROPERTIES;
}
Result<String> decodedUserName = null;
Result<byte[]> decodedPassword = null;
byte[] decodedPassword = null;
if (mqttConnectVariableHeader.hasUserName()) {
decodedUserName = decodeString(buffer);
numberOfBytesConsumed += decodedUserName.numberOfBytesConsumed;
}
if (mqttConnectVariableHeader.hasPassword()) {
decodedPassword = decodeByte(buffer);
numberOfBytesConsumed += decodedPassword.numberOfBytesConsumed;
decodedPassword = decodeByteArray(buffer);
numberOfBytesConsumed += decodedPassword.length + 2;
}
final MqttConnectPayload mqttConnectPayload =
new MqttConnectPayload(
decodedClientId.value,
willProperties,
decodedWillTopic != null ? decodedWillTopic.value : null,
decodedWillMessage != null ? decodedWillMessage.value : null,
decodedWillMessage,
decodedUserName != null ? decodedUserName.value : null,
decodedPassword != null ? decodedPassword.value : null);
decodedPassword);
return new Result<MqttConnectPayload>(mqttConnectPayload, numberOfBytesConsumed);
}
@@ -375,9 +578,21 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
final Result<String> decodedTopicName = decodeString(buffer);
numberOfBytesConsumed += decodedTopicName.numberOfBytesConsumed;
int qos = buffer.readUnsignedByte() & 0x03;
//See 3.8.3.1 Subscription Options of MQTT 5.0 specification for optionByte details
final short optionByte = buffer.readUnsignedByte();
MqttQoS qos = MqttQoS.valueOf(optionByte & 0x03);
boolean noLocal = ((optionByte & 0x04) >> 2) == 1;
boolean retainAsPublished = ((optionByte & 0x08) >> 3) == 1;
MqttSubscriptionOption.RetainedHandlingPolicy retainHandling = MqttSubscriptionOption.RetainedHandlingPolicy.valueOf((optionByte & 0x30) >> 4);
final MqttSubscriptionOption subscriptionOption = new MqttSubscriptionOption(qos,
noLocal,
retainAsPublished,
retainHandling);
numberOfBytesConsumed++;
subscribeTopics.add(new MqttTopicSubscription(decodedTopicName.value, MqttQoS.valueOf(qos)));
subscribeTopics.add(new MqttTopicSubscription(decodedTopicName.value, subscriptionOption));
}
return new Result<MqttSubscribePayload>(new MqttSubscribePayload(subscribeTopics), numberOfBytesConsumed);
}
@@ -385,16 +600,30 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
private static Result<MqttSubAckPayload> decodeSubackPayload(
ByteBuf buffer,
int bytesRemainingInVariablePart) {
final List<Integer> grantedQos = new ArrayList<Integer>();
final List<Integer> grantedQos = new ArrayList<Integer>(bytesRemainingInVariablePart);
int numberOfBytesConsumed = 0;
while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
int qos = buffer.readUnsignedByte() & 0x03;
int reasonCode = buffer.readUnsignedByte();
numberOfBytesConsumed++;
grantedQos.add(qos);
grantedQos.add(reasonCode);
}
return new Result<MqttSubAckPayload>(new MqttSubAckPayload(grantedQos), numberOfBytesConsumed);
}
private static Result<MqttUnsubAckPayload> decodeUnsubAckPayload(
ChannelHandlerContext ctx,
ByteBuf buffer,
int bytesRemainingInVariablePart) {
final List<Short> reasonCodes = new ArrayList<Short>(bytesRemainingInVariablePart);
int numberOfBytesConsumed = 0;
while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
short reasonCode = buffer.readUnsignedByte();
numberOfBytesConsumed++;
reasonCodes.add(reasonCode);
}
return new Result<MqttUnsubAckPayload>(new MqttUnsubAckPayload(reasonCodes), numberOfBytesConsumed);
}
private static Result<MqttUnsubscribePayload> decodeUnsubscribePayload(
ByteBuf buffer,
int bytesRemainingInVariablePart) {
@@ -419,25 +648,9 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
return decodeString(buffer, 0, Integer.MAX_VALUE);
}
private static Result<String> decodeAsciiString(ByteBuf buffer) {
Result<String> result = decodeString(buffer, 0, Integer.MAX_VALUE);
final String s = result.value;
for (int i = 0; i < s.length(); i++) {
if (s.charAt(i) > 127) {
return new Result<String>(null, result.numberOfBytesConsumed);
}
}
return new Result<String>(s, result.numberOfBytesConsumed);
}
private static Result<byte[]> decodeByte(ByteBuf buffer) {
return decodeByte(buffer, 0, Integer.MAX_VALUE);
}
private static Result<String> decodeString(ByteBuf buffer, int minBytes, int maxBytes) {
final Result<Integer> decodedSize = decodeMsbLsb(buffer);
int size = decodedSize.value;
int numberOfBytesConsumed = decodedSize.numberOfBytesConsumed;
int size = decodeMsbLsb(buffer);
int numberOfBytesConsumed = 2;
if (size < minBytes || size > maxBytes) {
buffer.skipBytes(size);
numberOfBytesConsumed += size;
@@ -449,35 +662,68 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
return new Result<String>(s, numberOfBytesConsumed);
}
private static Result<byte[]> decodeByte(ByteBuf buffer, int minBytes, int maxBytes) {
final Result<Integer> decodedSize = decodeMsbLsb(buffer);
int size = decodedSize.value;
int numberOfBytesConsumed = decodedSize.numberOfBytesConsumed;
if (size < minBytes || size > maxBytes) {
buffer.skipBytes(size);
numberOfBytesConsumed += size;
return new Result<>(null, numberOfBytesConsumed);
}
byte[] s = new byte[size];
buffer.getBytes(buffer.readerIndex(), s, 0, size);
buffer.skipBytes(size);
numberOfBytesConsumed += size;
return new Result<>(s, numberOfBytesConsumed);
/**
*
* @return the decoded byte[], numberOfBytesConsumed = byte[].length + 2
*/
private static byte[] decodeByteArray(ByteBuf buffer) {
int size = decodeMsbLsb(buffer);
byte[] bytes = new byte[size];
buffer.readBytes(bytes);
return bytes;
}
private static Result<Integer> decodeMsbLsb(ByteBuf buffer) {
return decodeMsbLsb(buffer, 0, 65535);
// packing utils to reduce the amount of garbage while decoding ints
private static long packInts(int a, int b) {
return (((long) a) << 32) | (b & 0xFFFFFFFFL);
}
private static Result<Integer> decodeMsbLsb(ByteBuf buffer, int min, int max) {
private static int unpackA(long ints) {
return (int) (ints >> 32);
}
private static int unpackB(long ints) {
return (int) ints;
}
/**
* numberOfBytesConsumed = 2. return decoded result.
*/
private static int decodeMsbLsb(ByteBuf buffer) {
int min = 0;
int max = 65535;
short msbSize = buffer.readUnsignedByte();
short lsbSize = buffer.readUnsignedByte();
final int numberOfBytesConsumed = 2;
int result = msbSize << 8 | lsbSize;
if (result < min || result > max) {
result = -1;
}
return new Result<Integer>(result, numberOfBytesConsumed);
return result;
}
/**
* See 1.5.5 Variable Byte Integer section of MQTT 5.0 specification for encoding/decoding rules
*
* @param buffer the buffer to decode from
* @return result pack with a = decoded integer, b = numberOfBytesConsumed. Need to unpack to read them.
* @throws DecoderException if bad MQTT protocol limits Remaining Length
*/
private static long decodeVariableByteInteger(ByteBuf buffer) {
int remainingLength = 0;
int multiplier = 1;
short digit;
int loops = 0;
do {
digit = buffer.readUnsignedByte();
remainingLength += (digit & 127) * multiplier;
multiplier *= 128;
loops++;
} while ((digit & 128) != 0 && loops < 4);
if (loops == 4 && (digit & 128) != 0) {
throw new DecoderException("MQTT protocol limits Remaining Length to 4 bytes");
}
return packInts(remainingLength, loops);
}
private static final class Result<T> {
@@ -490,4 +736,82 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
this.numberOfBytesConsumed = numberOfBytesConsumed;
}
}
private static Result<MqttProperties> decodeProperties(ByteBuf buffer) {
final long propertiesLength = decodeVariableByteInteger(buffer);
int totalPropertiesLength = unpackA(propertiesLength);
int numberOfBytesConsumed = unpackB(propertiesLength);
MqttProperties decodedProperties = new MqttProperties();
while (numberOfBytesConsumed < totalPropertiesLength) {
long propertyId = decodeVariableByteInteger(buffer);
final int propertyIdValue = unpackA(propertyId);
numberOfBytesConsumed += unpackB(propertyId);
MqttProperties.MqttPropertyType propertyType = MqttProperties.MqttPropertyType.valueOf(propertyIdValue);
switch (propertyType) {
case PAYLOAD_FORMAT_INDICATOR:
case REQUEST_PROBLEM_INFORMATION:
case REQUEST_RESPONSE_INFORMATION:
case MAXIMUM_QOS:
case RETAIN_AVAILABLE:
case WILDCARD_SUBSCRIPTION_AVAILABLE:
case SUBSCRIPTION_IDENTIFIER_AVAILABLE:
case SHARED_SUBSCRIPTION_AVAILABLE:
final int b1 = buffer.readUnsignedByte();
numberOfBytesConsumed++;
decodedProperties.add(new MqttProperties.IntegerProperty(propertyIdValue, b1));
break;
case SERVER_KEEP_ALIVE:
case RECEIVE_MAXIMUM:
case TOPIC_ALIAS_MAXIMUM:
case TOPIC_ALIAS:
final int int2BytesResult = decodeMsbLsb(buffer);
numberOfBytesConsumed += 2;
decodedProperties.add(new MqttProperties.IntegerProperty(propertyIdValue, int2BytesResult));
break;
case PUBLICATION_EXPIRY_INTERVAL:
case SESSION_EXPIRY_INTERVAL:
case WILL_DELAY_INTERVAL:
case MAXIMUM_PACKET_SIZE:
final int maxPacketSize = buffer.readInt();
numberOfBytesConsumed += 4;
decodedProperties.add(new MqttProperties.IntegerProperty(propertyIdValue, maxPacketSize));
break;
case SUBSCRIPTION_IDENTIFIER:
long vbIntegerResult = decodeVariableByteInteger(buffer);
numberOfBytesConsumed += unpackB(vbIntegerResult);
decodedProperties.add(new MqttProperties.IntegerProperty(propertyIdValue, unpackA(vbIntegerResult)));
break;
case CONTENT_TYPE:
case RESPONSE_TOPIC:
case ASSIGNED_CLIENT_IDENTIFIER:
case AUTHENTICATION_METHOD:
case RESPONSE_INFORMATION:
case SERVER_REFERENCE:
case REASON_STRING:
final Result<String> stringResult = decodeString(buffer);
numberOfBytesConsumed += stringResult.numberOfBytesConsumed;
decodedProperties.add(new MqttProperties.StringProperty(propertyIdValue, stringResult.value));
break;
case USER_PROPERTY:
final Result<String> keyResult = decodeString(buffer);
final Result<String> valueResult = decodeString(buffer);
numberOfBytesConsumed += keyResult.numberOfBytesConsumed;
numberOfBytesConsumed += valueResult.numberOfBytesConsumed;
decodedProperties.add(new MqttProperties.UserProperty(keyResult.value, valueResult.value));
break;
case CORRELATION_DATA:
case AUTHENTICATION_DATA:
final byte[] binaryDataResult = decodeByteArray(buffer);
numberOfBytesConsumed += binaryDataResult.length + 2;
decodedProperties.add(new MqttProperties.BinaryProperty(propertyIdValue, binaryDataResult));
break;
default:
//shouldn't reach here
throw new DecoderException("Unknown property type: " + propertyType);
}
}
return new Result<MqttProperties>(decodedProperties, numberOfBytesConsumed);
}
}
+517 -172
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -18,19 +18,22 @@ package io.netty.handler.codec.mqtt;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.EncoderException;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.EmptyArrays;
import java.util.List;
import static io.netty.handler.codec.mqtt.MqttCodecUtil.*;
import static io.netty.buffer.ByteBufUtil.*;
/**
* Encodes Mqtt messages into bytes following the protocol specification v3.1
* as described here <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html">MQTTV3.1</a>
* as described here <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html">MQTTV3.1</a>
* or v5.0 as described here <a href="https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html">MQTTv5.0</a> -
* depending on the version specified in the first CONNECT message that goes through the channel.
*/
@ChannelHandler.Sharable
public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
@@ -41,49 +44,57 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
@Override
protected void encode(ChannelHandlerContext ctx, MqttMessage msg, List<Object> out) throws Exception {
out.add(doEncode(ctx.alloc(), msg));
out.add(doEncode(ctx, msg));
}
/**
* This is the main encoding method.
* It's only visible for testing.
*
* @param byteBufAllocator Allocates ByteBuf
* @param message MQTT message to encode
* @return ByteBuf with encoded bytes
*/
static ByteBuf doEncode(ByteBufAllocator byteBufAllocator, MqttMessage message) {
static ByteBuf doEncode(ChannelHandlerContext ctx,
MqttMessage message) {
switch (message.fixedHeader().messageType()) {
case CONNECT:
return encodeConnectMessage(byteBufAllocator, (MqttConnectMessage) message);
return encodeConnectMessage(ctx, (MqttConnectMessage) message);
case CONNACK:
return encodeConnAckMessage(byteBufAllocator, (MqttConnAckMessage) message);
return encodeConnAckMessage(ctx, (MqttConnAckMessage) message);
case PUBLISH:
return encodePublishMessage(byteBufAllocator, (MqttPublishMessage) message);
return encodePublishMessage(ctx, (MqttPublishMessage) message);
case SUBSCRIBE:
return encodeSubscribeMessage(byteBufAllocator, (MqttSubscribeMessage) message);
return encodeSubscribeMessage(ctx, (MqttSubscribeMessage) message);
case UNSUBSCRIBE:
return encodeUnsubscribeMessage(byteBufAllocator, (MqttUnsubscribeMessage) message);
return encodeUnsubscribeMessage(ctx, (MqttUnsubscribeMessage) message);
case SUBACK:
return encodeSubAckMessage(byteBufAllocator, (MqttSubAckMessage) message);
return encodeSubAckMessage(ctx, (MqttSubAckMessage) message);
case UNSUBACK:
if (message instanceof MqttUnsubAckMessage) {
return encodeUnsubAckMessage(ctx, (MqttUnsubAckMessage) message);
}
return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(ctx.alloc(), message);
case PUBACK:
case PUBREC:
case PUBREL:
case PUBCOMP:
return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(byteBufAllocator, message);
return encodePubReplyMessage(ctx, message);
case DISCONNECT:
case AUTH:
return encodeReasonCodePlusPropertiesMessage(ctx, message);
case PINGREQ:
case PINGRESP:
case DISCONNECT:
return encodeMessageWithOnlySingleByteFixedHeader(byteBufAllocator, message);
return encodeMessageWithOnlySingleByteFixedHeader(ctx.alloc(), message);
default:
throw new IllegalArgumentException(
@@ -92,7 +103,7 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
}
private static ByteBuf encodeConnectMessage(
ByteBufAllocator byteBufAllocator,
ChannelHandlerContext ctx,
MqttConnectMessage message) {
int payloadBufferSize = 0;
@@ -101,70 +112,96 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
MqttConnectPayload payload = message.payload();
MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(variableHeader.name(),
(byte) variableHeader.version());
MqttCodecUtil.setMqttVersion(ctx, mqttVersion);
// as MQTT 3.1 & 3.1.1 spec, If the User Name Flag is set to 0, the Password Flag MUST be set to 0
if (!variableHeader.hasUserName() && variableHeader.hasPassword()) {
throw new EncoderException("Without a username, the password MUST be not set");
}
// Client id
String clientIdentifier = payload.clientIdentifier();
if (!isValidClientId(mqttVersion, clientIdentifier)) {
if (!MqttCodecUtil.isValidClientId(mqttVersion, MqttConstant.DEFAULT_MAX_CLIENT_ID_LENGTH, clientIdentifier)) {
throw new MqttIdentifierRejectedException("invalid clientIdentifier: " + clientIdentifier);
}
byte[] clientIdentifierBytes = encodeStringUtf8(clientIdentifier);
payloadBufferSize += 2 + clientIdentifierBytes.length;
int clientIdentifierBytes = utf8Bytes(clientIdentifier);
payloadBufferSize += 2 + clientIdentifierBytes;
// Will topic and message
String willTopic = payload.willTopic();
byte[] willTopicBytes = willTopic != null ? encodeStringUtf8(willTopic) : EmptyArrays.EMPTY_BYTES;
String willMessage = payload.willMessage();
byte[] willMessageBytes = willMessage != null ? encodeStringUtf8(willMessage) : EmptyArrays.EMPTY_BYTES;
int willTopicBytes = nullableUtf8Bytes(willTopic);
byte[] willMessage = payload.willMessageInBytes();
byte[] willMessageBytes = willMessage != null ? willMessage : EmptyArrays.EMPTY_BYTES;
if (variableHeader.isWillFlag()) {
payloadBufferSize += 2 + willTopicBytes.length;
payloadBufferSize += 2 + willTopicBytes;
payloadBufferSize += 2 + willMessageBytes.length;
}
String userName = payload.userName();
byte[] userNameBytes = userName != null ? encodeStringUtf8(userName) : EmptyArrays.EMPTY_BYTES;
int userNameBytes = nullableUtf8Bytes(userName);
if (variableHeader.hasUserName()) {
payloadBufferSize += 2 + userNameBytes.length;
payloadBufferSize += 2 + userNameBytes;
}
byte[] passwordBytes = payload.password();
byte[] password = payload.passwordInBytes();
byte[] passwordBytes = password != null ? password : EmptyArrays.EMPTY_BYTES;
if (variableHeader.hasPassword()) {
payloadBufferSize += 2 + passwordBytes.length;
}
// Fixed header
// Fixed and variable header
byte[] protocolNameBytes = mqttVersion.protocolNameBytes();
int variableHeaderBufferSize = 2 + protocolNameBytes.length + 4;
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = byteBufAllocator.buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
ByteBuf propertiesBuf = encodePropertiesIfNeeded(
mqttVersion,
ctx.alloc(),
message.variableHeader().properties());
try {
final ByteBuf willPropertiesBuf;
if (variableHeader.isWillFlag()) {
willPropertiesBuf = encodePropertiesIfNeeded(mqttVersion, ctx.alloc(), payload.willProperties());
payloadBufferSize += willPropertiesBuf.readableBytes();
} else {
willPropertiesBuf = Unpooled.EMPTY_BUFFER;
}
try {
int variableHeaderBufferSize = 2 + protocolNameBytes.length + 4 + propertiesBuf.readableBytes();
buf.writeShort(protocolNameBytes.length);
buf.writeBytes(protocolNameBytes);
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
buf.writeByte(variableHeader.version());
buf.writeByte(getConnVariableHeaderFlag(variableHeader));
buf.writeShort(variableHeader.keepAliveTimeSeconds());
buf.writeShort(protocolNameBytes.length);
buf.writeBytes(protocolNameBytes);
// Payload
buf.writeShort(clientIdentifierBytes.length);
buf.writeBytes(clientIdentifierBytes, 0, clientIdentifierBytes.length);
if (variableHeader.isWillFlag()) {
buf.writeShort(willTopicBytes.length);
buf.writeBytes(willTopicBytes, 0, willTopicBytes.length);
buf.writeShort(willMessageBytes.length);
buf.writeBytes(willMessageBytes, 0, willMessageBytes.length);
buf.writeByte(variableHeader.version());
buf.writeByte(getConnVariableHeaderFlag(variableHeader));
buf.writeShort(variableHeader.keepAliveTimeSeconds());
buf.writeBytes(propertiesBuf);
// Payload
writeExactUTF8String(buf, clientIdentifier, clientIdentifierBytes);
if (variableHeader.isWillFlag()) {
buf.writeBytes(willPropertiesBuf);
writeExactUTF8String(buf, willTopic, willTopicBytes);
buf.writeShort(willMessageBytes.length);
buf.writeBytes(willMessageBytes, 0, willMessageBytes.length);
}
if (variableHeader.hasUserName()) {
writeExactUTF8String(buf, userName, userNameBytes);
}
if (variableHeader.hasPassword()) {
buf.writeShort(passwordBytes.length);
buf.writeBytes(passwordBytes, 0, passwordBytes.length);
}
return buf;
} finally {
willPropertiesBuf.release();
}
} finally {
propertiesBuf.release();
}
if (variableHeader.hasUserName()) {
buf.writeShort(userNameBytes.length);
buf.writeBytes(userNameBytes, 0, userNameBytes.length);
}
if (variableHeader.hasPassword()) {
buf.writeShort(passwordBytes.length);
buf.writeBytes(passwordBytes, 0, passwordBytes.length);
}
return buf;
}
private static int getConnVariableHeaderFlag(MqttConnectVariableHeader variableHeader) {
@@ -189,149 +226,285 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
}
private static ByteBuf encodeConnAckMessage(
ByteBufAllocator byteBufAllocator,
ChannelHandlerContext ctx,
MqttConnAckMessage message) {
int length = 4;
MqttConnectAckPayload payload = null;
if (message.payload() != null) {
payload = (MqttConnectAckPayload)message.payload();
if (payload.getData() != null && payload.getData().length > 0) {
length += payload.getData().length;
}
}
ByteBuf buf = byteBufAllocator.buffer(length);
buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
buf.writeByte(length - 2);
buf.writeByte(message.variableHeader().isSessionPresent() ? 0x01 : 0x00);
buf.writeByte(message.variableHeader().connectReturnCode().byteValue());
if (length > 4) {
buf.writeBytes(payload.getData());
}
final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
ctx.alloc(),
message.variableHeader().properties());
return buf;
try {
int length = 4;
MqttConnectAckPayload payload = null;
if (message.payload() != null) {
payload = (MqttConnectAckPayload)message.payload();
if (payload.getData() != null && payload.getData().length > 0) {
length += payload.getData().length;
}
}
ByteBuf buf = ctx.alloc().buffer(length + propertiesBuf.readableBytes());
buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
writeVariableLengthInt(buf, length - 2 + propertiesBuf.readableBytes());
buf.writeByte(message.variableHeader().isSessionPresent() ? 0x01 : 0x00);
buf.writeByte(message.variableHeader().connectReturnCode().byteValue());
if (length > 4) {
buf.writeBytes(payload.getData());
}
buf.writeBytes(propertiesBuf);
return buf;
} finally {
propertiesBuf.release();
}
}
private static ByteBuf encodeSubscribeMessage(
ByteBufAllocator byteBufAllocator,
ChannelHandlerContext ctx,
MqttSubscribeMessage message) {
int variableHeaderBufferSize = 2;
int payloadBufferSize = 0;
MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
ctx.alloc(),
message.idAndPropertiesVariableHeader().properties());
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttMessageIdVariableHeader variableHeader = message.variableHeader();
MqttSubscribePayload payload = message.payload();
try {
final int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
int payloadBufferSize = 0;
for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
String topicName = topic.topicName();
byte[] topicNameBytes = encodeStringUtf8(topicName);
payloadBufferSize += 2 + topicNameBytes.length;
payloadBufferSize += 1;
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttMessageIdVariableHeader variableHeader = message.variableHeader();
MqttSubscribePayload payload = message.payload();
for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
String topicName = topic.topicName();
int topicNameBytes = utf8Bytes(topicName);
payloadBufferSize += 2 + topicNameBytes;
payloadBufferSize += 1;
}
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
// Variable Header
int messageId = variableHeader.messageId();
buf.writeShort(messageId);
buf.writeBytes(propertiesBuf);
// Payload
for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
writeUnsafeUTF8String(buf, topic.topicName());
if (mqttVersion.protocolLevel() >= MqttVersion.MQTT_3_1_1.protocolLevel()) {
buf.writeByte(topic.qualityOfService().value());
} else {
final MqttSubscriptionOption option = topic.option();
int optionEncoded = option.retainHandling().value() << 4;
if (option.isRetainAsPublished()) {
optionEncoded |= 0x08;
}
if (option.isNoLocal()) {
optionEncoded |= 0x04;
}
optionEncoded |= option.qos().value();
buf.writeByte(optionEncoded);
}
}
return buf;
} finally {
propertiesBuf.release();
}
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = byteBufAllocator.buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
// Variable Header
int messageId = variableHeader.messageId();
buf.writeShort(messageId);
// Payload
for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
String topicName = topic.topicName();
byte[] topicNameBytes = encodeStringUtf8(topicName);
buf.writeShort(topicNameBytes.length);
buf.writeBytes(topicNameBytes, 0, topicNameBytes.length);
buf.writeByte(topic.qualityOfService().value());
}
return buf;
}
private static ByteBuf encodeUnsubscribeMessage(
ByteBufAllocator byteBufAllocator,
ChannelHandlerContext ctx,
MqttUnsubscribeMessage message) {
int variableHeaderBufferSize = 2;
int payloadBufferSize = 0;
MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
ctx.alloc(),
message.idAndPropertiesVariableHeader().properties());
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttMessageIdVariableHeader variableHeader = message.variableHeader();
MqttUnsubscribePayload payload = message.payload();
try {
final int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
int payloadBufferSize = 0;
for (String topicName : payload.topics()) {
byte[] topicNameBytes = encodeStringUtf8(topicName);
payloadBufferSize += 2 + topicNameBytes.length;
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttMessageIdVariableHeader variableHeader = message.variableHeader();
MqttUnsubscribePayload payload = message.payload();
for (String topicName : payload.topics()) {
int topicNameBytes = utf8Bytes(topicName);
payloadBufferSize += 2 + topicNameBytes;
}
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
// Variable Header
int messageId = variableHeader.messageId();
buf.writeShort(messageId);
buf.writeBytes(propertiesBuf);
// Payload
for (String topicName : payload.topics()) {
writeUnsafeUTF8String(buf, topicName);
}
return buf;
} finally {
propertiesBuf.release();
}
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = byteBufAllocator.buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
// Variable Header
int messageId = variableHeader.messageId();
buf.writeShort(messageId);
// Payload
for (String topicName : payload.topics()) {
byte[] topicNameBytes = encodeStringUtf8(topicName);
buf.writeShort(topicNameBytes.length);
buf.writeBytes(topicNameBytes, 0, topicNameBytes.length);
}
return buf;
}
private static ByteBuf encodeSubAckMessage(
ByteBufAllocator byteBufAllocator,
ChannelHandlerContext ctx,
MqttSubAckMessage message) {
int variableHeaderBufferSize = 2;
int payloadBufferSize = message.payload().grantedQoSLevels().size();
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = byteBufAllocator.buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
writeVariableLengthInt(buf, variablePartSize);
buf.writeShort(message.variableHeader().messageId());
for (int qos : message.payload().grantedQoSLevels()) {
buf.writeByte(qos);
}
MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
ctx.alloc(),
message.idAndPropertiesVariableHeader().properties());
try {
int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
int payloadBufferSize = message.payload().grantedQoSLevels().size();
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
writeVariableLengthInt(buf, variablePartSize);
buf.writeShort(message.variableHeader().messageId());
buf.writeBytes(propertiesBuf);
for (int code: message.payload().reasonCodes()) {
buf.writeByte(code);
}
return buf;
return buf;
} finally {
propertiesBuf.release();
}
}
private static ByteBuf encodeUnsubAckMessage(
ChannelHandlerContext ctx,
MqttUnsubAckMessage message) {
if (message.variableHeader() instanceof MqttMessageIdAndPropertiesVariableHeader) {
MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
ctx.alloc(),
message.idAndPropertiesVariableHeader().properties());
try {
int variableHeaderBufferSize = 2 + propertiesBuf.readableBytes();
MqttUnsubAckPayload payload = message.payload();
int payloadBufferSize = payload == null ? 0 : payload.unsubscribeReasonCodes().size();
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(message.fixedHeader()));
writeVariableLengthInt(buf, variablePartSize);
buf.writeShort(message.variableHeader().messageId());
buf.writeBytes(propertiesBuf);
if (payload != null) {
for (Short reasonCode : payload.unsubscribeReasonCodes()) {
buf.writeByte(reasonCode);
}
}
return buf;
} finally {
propertiesBuf.release();
}
} else {
return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(ctx.alloc(), message);
}
}
private static ByteBuf encodePublishMessage(
ByteBufAllocator byteBufAllocator,
ChannelHandlerContext ctx,
MqttPublishMessage message) {
MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttPublishVariableHeader variableHeader = message.variableHeader();
ByteBuf payload = message.payload().duplicate();
String topicName = variableHeader.topicName();
byte[] topicNameBytes = encodeStringUtf8(topicName);
int topicNameBytes = utf8Bytes(topicName);
int variableHeaderBufferSize = 2 + topicNameBytes.length +
(mqttFixedHeader.qosLevel().value() > 0 ? 2 : 0);
int payloadBufferSize = payload.readableBytes();
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf propertiesBuf = encodePropertiesIfNeeded(mqttVersion,
ctx.alloc(),
message.variableHeader().properties());
ByteBuf buf = byteBufAllocator.buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
buf.writeShort(topicNameBytes.length);
buf.writeBytes(topicNameBytes);
if (mqttFixedHeader.qosLevel().value() > 0) {
buf.writeShort(variableHeader.messageId());
try {
int variableHeaderBufferSize = 2 + topicNameBytes +
(mqttFixedHeader.qosLevel().value() > 0 ? 2 : 0) + propertiesBuf.readableBytes();
int payloadBufferSize = payload.readableBytes();
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variablePartSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
writeExactUTF8String(buf, topicName, topicNameBytes);
if (mqttFixedHeader.qosLevel().value() > 0) {
buf.writeShort(variableHeader.packetId());
}
buf.writeBytes(propertiesBuf);
buf.writeBytes(payload);
return buf;
} finally {
propertiesBuf.release();
}
buf.writeBytes(payload);
}
return buf;
private static ByteBuf encodePubReplyMessage(ChannelHandlerContext ctx,
MqttMessage message) {
if (message.variableHeader() instanceof MqttPubReplyMessageVariableHeader) {
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttPubReplyMessageVariableHeader variableHeader =
(MqttPubReplyMessageVariableHeader) message.variableHeader();
int msgId = variableHeader.messageId();
final ByteBuf propertiesBuf;
final boolean includeReasonCode;
final int variableHeaderBufferSize;
final MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
if (mqttVersion == MqttVersion.MQTT_5 &&
(variableHeader.reasonCode() != MqttPubReplyMessageVariableHeader.REASON_CODE_OK ||
!variableHeader.properties().isEmpty())) {
propertiesBuf = encodeProperties(ctx.alloc(), variableHeader.properties());
includeReasonCode = true;
variableHeaderBufferSize = 3 + propertiesBuf.readableBytes();
} else {
propertiesBuf = Unpooled.EMPTY_BUFFER;
includeReasonCode = false;
variableHeaderBufferSize = 2;
}
try {
final int fixedHeaderBufferSize = 1 + getVariableLengthInt(variableHeaderBufferSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variableHeaderBufferSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variableHeaderBufferSize);
buf.writeShort(msgId);
if (includeReasonCode) {
buf.writeByte(variableHeader.reasonCode());
}
buf.writeBytes(propertiesBuf);
return buf;
} finally {
propertiesBuf.release();
}
} else {
return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(ctx.alloc(), message);
}
}
private static ByteBuf encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(
@@ -342,12 +515,11 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
int msgId = variableHeader.messageId();
int variableHeaderBufferSize = 2; // variable part only has a message id
ByteBuf payload = null;
if (message.payload() != null) {
payload = ((ByteBuf)message.payload()).duplicate();
variableHeaderBufferSize += payload.readableBytes();
}
}
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variableHeaderBufferSize);
ByteBuf buf = byteBufAllocator.buffer(fixedHeaderBufferSize + variableHeaderBufferSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
@@ -355,11 +527,53 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
buf.writeShort(msgId);
if (payload != null) {
buf.writeBytes(payload);
}
}
return buf;
}
private static ByteBuf encodeReasonCodePlusPropertiesMessage(
ChannelHandlerContext ctx,
MqttMessage message) {
if (message.variableHeader() instanceof MqttReasonCodeAndPropertiesVariableHeader) {
MqttVersion mqttVersion = MqttCodecUtil.getMqttVersion(ctx);
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttReasonCodeAndPropertiesVariableHeader variableHeader =
(MqttReasonCodeAndPropertiesVariableHeader) message.variableHeader();
final ByteBuf propertiesBuf;
final boolean includeReasonCode;
final int variableHeaderBufferSize;
if (mqttVersion == MqttVersion.MQTT_5 &&
(variableHeader.reasonCode() != MqttReasonCodeAndPropertiesVariableHeader.REASON_CODE_OK ||
!variableHeader.properties().isEmpty())) {
propertiesBuf = encodeProperties(ctx.alloc(), variableHeader.properties());
includeReasonCode = true;
variableHeaderBufferSize = 1 + propertiesBuf.readableBytes();
} else {
propertiesBuf = Unpooled.EMPTY_BUFFER;
includeReasonCode = false;
variableHeaderBufferSize = 0;
}
try {
final int fixedHeaderBufferSize = 1 + getVariableLengthInt(variableHeaderBufferSize);
ByteBuf buf = ctx.alloc().buffer(fixedHeaderBufferSize + variableHeaderBufferSize);
buf.writeByte(getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variableHeaderBufferSize);
if (includeReasonCode) {
buf.writeByte(variableHeader.reasonCode());
}
buf.writeBytes(propertiesBuf);
return buf;
} finally {
propertiesBuf.release();
}
} else {
return encodeMessageWithOnlySingleByteFixedHeader(ctx.alloc(), message);
}
}
private static ByteBuf encodeMessageWithOnlySingleByteFixedHeader(
ByteBufAllocator byteBufAllocator,
MqttMessage message) {
@@ -371,6 +585,104 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
return buf;
}
private static ByteBuf encodePropertiesIfNeeded(MqttVersion mqttVersion,
ByteBufAllocator byteBufAllocator,
MqttProperties mqttProperties) {
if (mqttVersion == MqttVersion.MQTT_5) {
return encodeProperties(byteBufAllocator, mqttProperties);
}
return Unpooled.EMPTY_BUFFER;
}
private static ByteBuf encodeProperties(ByteBufAllocator byteBufAllocator,
MqttProperties mqttProperties) {
ByteBuf propertiesHeaderBuf = byteBufAllocator.buffer();
// encode also the Properties part
try {
ByteBuf propertiesBuf = byteBufAllocator.buffer();
try {
for (MqttProperties.MqttProperty property : mqttProperties.listAll()) {
MqttProperties.MqttPropertyType propertyType =
MqttProperties.MqttPropertyType.valueOf(property.propertyId);
switch (propertyType) {
case PAYLOAD_FORMAT_INDICATOR:
case REQUEST_PROBLEM_INFORMATION:
case REQUEST_RESPONSE_INFORMATION:
case MAXIMUM_QOS:
case RETAIN_AVAILABLE:
case WILDCARD_SUBSCRIPTION_AVAILABLE:
case SUBSCRIPTION_IDENTIFIER_AVAILABLE:
case SHARED_SUBSCRIPTION_AVAILABLE:
writeVariableLengthInt(propertiesBuf, property.propertyId);
final byte bytePropValue = ((MqttProperties.IntegerProperty) property).value.byteValue();
propertiesBuf.writeByte(bytePropValue);
break;
case SERVER_KEEP_ALIVE:
case RECEIVE_MAXIMUM:
case TOPIC_ALIAS_MAXIMUM:
case TOPIC_ALIAS:
writeVariableLengthInt(propertiesBuf, property.propertyId);
final short twoBytesInPropValue =
((MqttProperties.IntegerProperty) property).value.shortValue();
propertiesBuf.writeShort(twoBytesInPropValue);
break;
case PUBLICATION_EXPIRY_INTERVAL:
case SESSION_EXPIRY_INTERVAL:
case WILL_DELAY_INTERVAL:
case MAXIMUM_PACKET_SIZE:
writeVariableLengthInt(propertiesBuf, property.propertyId);
final int fourBytesIntPropValue = ((MqttProperties.IntegerProperty) property).value;
propertiesBuf.writeInt(fourBytesIntPropValue);
break;
case SUBSCRIPTION_IDENTIFIER:
writeVariableLengthInt(propertiesBuf, property.propertyId);
final int vbi = ((MqttProperties.IntegerProperty) property).value;
writeVariableLengthInt(propertiesBuf, vbi);
break;
case CONTENT_TYPE:
case RESPONSE_TOPIC:
case ASSIGNED_CLIENT_IDENTIFIER:
case AUTHENTICATION_METHOD:
case RESPONSE_INFORMATION:
case SERVER_REFERENCE:
case REASON_STRING:
writeVariableLengthInt(propertiesBuf, property.propertyId);
writeEagerUTF8String(propertiesBuf, ((MqttProperties.StringProperty) property).value);
break;
case USER_PROPERTY:
final List<MqttProperties.StringPair> pairs =
((MqttProperties.UserProperties) property).value;
for (MqttProperties.StringPair pair : pairs) {
writeVariableLengthInt(propertiesBuf, property.propertyId);
writeEagerUTF8String(propertiesBuf, pair.key);
writeEagerUTF8String(propertiesBuf, pair.value);
}
break;
case CORRELATION_DATA:
case AUTHENTICATION_DATA:
writeVariableLengthInt(propertiesBuf, property.propertyId);
final byte[] binaryPropValue = ((MqttProperties.BinaryProperty) property).value;
propertiesBuf.writeShort(binaryPropValue.length);
propertiesBuf.writeBytes(binaryPropValue, 0, binaryPropValue.length);
break;
default:
//shouldn't reach here
throw new EncoderException("Unknown property type: " + propertyType);
}
}
writeVariableLengthInt(propertiesHeaderBuf, propertiesBuf.readableBytes());
propertiesHeaderBuf.writeBytes(propertiesBuf);
return propertiesHeaderBuf;
} finally {
propertiesBuf.release();
}
} catch (RuntimeException e) {
propertiesHeaderBuf.release();
throw e;
}
}
private static int getFixedHeaderByte1(MqttFixedHeader header) {
int ret = 0;
ret |= header.messageType().value() << 4;
@@ -395,6 +707,42 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
} while (num > 0);
}
private static int nullableUtf8Bytes(String s) {
return s == null? 0 : utf8Bytes(s);
}
private static int nullableMaxUtf8Bytes(String s) {
return s == null? 0 : utf8MaxBytes(s);
}
private static void writeExactUTF8String(ByteBuf buf, String s, int utf8Length) {
buf.ensureWritable(utf8Length + 2);
buf.writeShort(utf8Length);
if (utf8Length > 0) {
final int writtenUtf8Length = reserveAndWriteUtf8(buf, s, utf8Length);
assert writtenUtf8Length == utf8Length;
}
}
private static void writeEagerUTF8String(ByteBuf buf, String s) {
final int maxUtf8Length = nullableMaxUtf8Bytes(s);
buf.ensureWritable(maxUtf8Length + 2);
final int writerIndex = buf.writerIndex();
final int startUtf8String = writerIndex + 2;
buf.writerIndex(startUtf8String);
final int utf8Length = s != null? reserveAndWriteUtf8(buf, s, maxUtf8Length) : 0;
buf.setShort(writerIndex, utf8Length);
}
private static void writeUnsafeUTF8String(ByteBuf buf, String s) {
final int writerIndex = buf.writerIndex();
final int startUtf8String = writerIndex + 2;
// no need to reserve any capacity here, already done earlier: that's why is Unsafe
buf.writerIndex(startUtf8String);
final int utf8Length = s != null? reserveAndWriteUtf8(buf, s, 0) : 0;
buf.setShort(writerIndex, utf8Length);
}
private static int getVariableLengthInt(int num) {
int count = 0;
do {
@@ -404,7 +752,4 @@ public final class MqttEncoder extends MessageToMessageEncoder<MqttMessage> {
return count;
}
private static byte[] encodeStringUtf8(String s) {
return s.getBytes(CharsetUtil.UTF_8);
}
}
+2 -2
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -20,7 +20,7 @@ import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.StringUtil;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#fixed-header">
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#fixed-header">
* MQTTV3.1/fixed-header</a>
*/
public final class MqttFixedHeader {
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+12 -1
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -29,6 +29,17 @@ public class MqttMessage {
private final Object payload;
private final DecoderResult decoderResult;
// Constants for fixed-header only message types with all flags set to 0 (see
// https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Table_2.2_-)
public static final MqttMessage PINGREQ = new MqttMessage(new MqttFixedHeader(MqttMessageType.PINGREQ, false,
MqttQoS.AT_MOST_ONCE, false, 0));
public static final MqttMessage PINGRESP = new MqttMessage(new MqttFixedHeader(MqttMessageType.PINGRESP, false,
MqttQoS.AT_MOST_ONCE, false, 0));
public static final MqttMessage DISCONNECT = new MqttMessage(new MqttFixedHeader(MqttMessageType.DISCONNECT, false,
MqttQoS.AT_MOST_ONCE, false, 0));
public MqttMessage(MqttFixedHeader mqttFixedHeader) {
this(mqttFixedHeader, null, null);
}
+493 -14
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -15,10 +15,14 @@
*/
package io.netty.handler.codec.mqtt;
import static io.netty.util.internal.ObjectUtil.checkPositive;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public final class MqttMessageBuilders {
@@ -29,6 +33,7 @@ public final class MqttMessageBuilders {
private MqttQoS qos;
private ByteBuf payload;
private int messageId;
private MqttProperties mqttProperties;
PublishBuilder() {
}
@@ -58,9 +63,15 @@ public final class MqttMessageBuilders {
return this;
}
public PublishBuilder properties(MqttProperties properties) {
this.mqttProperties = properties;
return this;
}
public MqttPublishMessage build() {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0);
MqttPublishVariableHeader mqttVariableHeader = new MqttPublishVariableHeader(topic, messageId);
MqttPublishVariableHeader mqttVariableHeader =
new MqttPublishVariableHeader(topic, messageId, mqttProperties);
return new MqttPublishMessage(mqttFixedHeader, mqttVariableHeader, Unpooled.buffer().writeBytes(payload));
}
}
@@ -73,13 +84,15 @@ public final class MqttMessageBuilders {
private boolean hasUser;
private boolean hasPassword;
private int keepAliveSecs;
private MqttProperties willProperties = MqttProperties.NO_PROPERTIES;
private boolean willFlag;
private boolean willRetain;
private MqttQoS willQos = MqttQoS.AT_MOST_ONCE;
private String willTopic;
private String willMessage;
private byte[] willMessage;
private String username;
private byte[] password;
private MqttProperties properties = MqttProperties.NO_PROPERTIES;
ConnectBuilder() {
}
@@ -119,7 +132,16 @@ public final class MqttMessageBuilders {
return this;
}
/**
* @deprecated use {@link ConnectBuilder#willMessage(byte[])} instead
*/
@Deprecated
public ConnectBuilder willMessage(String willMessage) {
willMessage(willMessage == null ? null : willMessage.getBytes(CharsetUtil.UTF_8));
return this;
}
public ConnectBuilder willMessage(byte[] willMessage) {
this.willMessage = willMessage;
return this;
}
@@ -129,6 +151,11 @@ public final class MqttMessageBuilders {
return this;
}
public ConnectBuilder willProperties(MqttProperties willProperties) {
this.willProperties = willProperties;
return this;
}
public ConnectBuilder hasUser(boolean value) {
this.hasUser = value;
return this;
@@ -140,17 +167,31 @@ public final class MqttMessageBuilders {
}
public ConnectBuilder username(String username) {
this.hasUser = true;
this.hasUser = username != null;
this.username = username;
return this;
}
/**
* @deprecated use {@link ConnectBuilder#password(byte[])} instead
*/
@Deprecated
public ConnectBuilder password(String password) {
password(password == null ? null : password.getBytes(CharsetUtil.UTF_8));
return this;
}
public ConnectBuilder password(byte[] password) {
this.hasPassword = true;
this.hasPassword = password != null;
this.password = password;
return this;
}
public ConnectBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public MqttConnectMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
@@ -164,9 +205,10 @@ public final class MqttMessageBuilders {
willQos.value(),
willFlag,
cleanSession,
keepAliveSecs);
keepAliveSecs,
properties);
MqttConnectPayload mqttConnectPayload =
new MqttConnectPayload(clientId, willTopic, willMessage, username, password);
new MqttConnectPayload(clientId, willProperties, willTopic, willMessage, username, password);
return new MqttConnectMessage(mqttFixedHeader, mqttConnectVariableHeader, mqttConnectPayload);
}
}
@@ -175,36 +217,54 @@ public final class MqttMessageBuilders {
private List<MqttTopicSubscription> subscriptions;
private int messageId;
private MqttProperties properties;
SubscribeBuilder() {
}
public SubscribeBuilder addSubscription(MqttQoS qos, String topic) {
if (subscriptions == null) {
subscriptions = new ArrayList<MqttTopicSubscription>(5);
}
ensureSubscriptionsExist();
subscriptions.add(new MqttTopicSubscription(topic, qos));
return this;
}
public SubscribeBuilder addSubscription(String topic, MqttSubscriptionOption option) {
ensureSubscriptionsExist();
subscriptions.add(new MqttTopicSubscription(topic, option));
return this;
}
public SubscribeBuilder messageId(int messageId) {
this.messageId = messageId;
return this;
}
public SubscribeBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public MqttSubscribeMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId);
MqttMessageIdAndPropertiesVariableHeader mqttVariableHeader =
new MqttMessageIdAndPropertiesVariableHeader(messageId, properties);
MqttSubscribePayload mqttSubscribePayload = new MqttSubscribePayload(subscriptions);
return new MqttSubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload);
}
private void ensureSubscriptionsExist() {
if (subscriptions == null) {
subscriptions = new ArrayList<MqttTopicSubscription>(5);
}
}
}
public static final class UnsubscribeBuilder {
private List<String> topicFilters;
private int messageId;
private MqttProperties properties;
UnsubscribeBuilder() {
}
@@ -222,21 +282,33 @@ public final class MqttMessageBuilders {
return this;
}
public UnsubscribeBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public MqttUnsubscribeMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.UNSUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId);
MqttMessageIdAndPropertiesVariableHeader mqttVariableHeader =
new MqttMessageIdAndPropertiesVariableHeader(messageId, properties);
MqttUnsubscribePayload mqttSubscribePayload = new MqttUnsubscribePayload(topicFilters);
return new MqttUnsubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload);
}
}
public interface PropertiesInitializer<T> {
void apply(T builder);
}
public static final class ConnAckBuilder {
private MqttConnectReturnCode returnCode;
private boolean sessionPresent;
private MqttProperties properties = MqttProperties.NO_PROPERTIES;
private ConnAckPropertiesBuilder propsBuilder;
ConnAckBuilder() {
private ConnAckBuilder() {
}
public ConnAckBuilder returnCode(MqttConnectReturnCode returnCode) {
@@ -249,15 +321,402 @@ public final class MqttMessageBuilders {
return this;
}
public ConnAckBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public ConnAckBuilder properties(PropertiesInitializer<ConnAckPropertiesBuilder> consumer) {
if (propsBuilder == null) {
propsBuilder = new ConnAckPropertiesBuilder();
}
consumer.apply(propsBuilder);
return this;
}
public MqttConnAckMessage build() {
if (propsBuilder != null) {
properties = propsBuilder.build();
}
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttConnAckVariableHeader mqttConnAckVariableHeader =
new MqttConnAckVariableHeader(returnCode, sessionPresent);
new MqttConnAckVariableHeader(returnCode, sessionPresent, properties);
return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}
}
public static final class ConnAckPropertiesBuilder {
private String clientId;
private Long sessionExpiryInterval;
private int receiveMaximum;
private Byte maximumQos;
private boolean retain;
private Long maximumPacketSize;
private int topicAliasMaximum;
private String reasonString;
private final MqttProperties.UserProperties userProperties = new MqttProperties.UserProperties();
private Boolean wildcardSubscriptionAvailable;
private Boolean subscriptionIdentifiersAvailable;
private Boolean sharedSubscriptionAvailable;
private Integer serverKeepAlive;
private String responseInformation;
private String serverReference;
private String authenticationMethod;
private byte[] authenticationData;
public MqttProperties build() {
final MqttProperties props = new MqttProperties();
if (clientId != null) {
props.add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.ASSIGNED_CLIENT_IDENTIFIER.value(),
clientId));
}
if (sessionExpiryInterval != null) {
props.add(new MqttProperties.IntegerProperty(
MqttProperties.MqttPropertyType.SESSION_EXPIRY_INTERVAL.value(), sessionExpiryInterval.intValue()));
}
if (receiveMaximum > 0) {
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.RECEIVE_MAXIMUM.value(), receiveMaximum));
}
if (maximumQos != null) {
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.MAXIMUM_QOS.value(), receiveMaximum));
}
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.RETAIN_AVAILABLE.value(), retain ? 1 : 0));
if (maximumPacketSize != null) {
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.MAXIMUM_PACKET_SIZE.value(),
maximumPacketSize.intValue()));
}
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.TOPIC_ALIAS_MAXIMUM.value(),
topicAliasMaximum));
if (reasonString != null) {
props.add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.REASON_STRING.value(), reasonString));
}
props.add(userProperties);
if (wildcardSubscriptionAvailable != null) {
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.WILDCARD_SUBSCRIPTION_AVAILABLE.value(),
wildcardSubscriptionAvailable ? 1 : 0));
}
if (subscriptionIdentifiersAvailable != null) {
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.SUBSCRIPTION_IDENTIFIER_AVAILABLE.value(),
subscriptionIdentifiersAvailable ? 1 : 0));
}
if (sharedSubscriptionAvailable != null) {
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.SHARED_SUBSCRIPTION_AVAILABLE.value(),
sharedSubscriptionAvailable ? 1 : 0));
}
if (serverKeepAlive != null) {
props.add(new MqttProperties.IntegerProperty(MqttProperties.MqttPropertyType.SERVER_KEEP_ALIVE.value(),
serverKeepAlive));
}
if (responseInformation != null) {
props.add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.RESPONSE_INFORMATION.value(),
responseInformation));
}
if (serverReference != null) {
props.add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.SERVER_REFERENCE.value(),
serverReference));
}
if (authenticationMethod != null) {
props.add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.AUTHENTICATION_METHOD.value(),
authenticationMethod));
}
if (authenticationData != null) {
props.add(new MqttProperties.BinaryProperty(MqttProperties.MqttPropertyType.AUTHENTICATION_DATA.value(),
authenticationData));
}
return props;
}
public ConnAckPropertiesBuilder sessionExpiryInterval(long seconds) {
this.sessionExpiryInterval = seconds;
return this;
}
public ConnAckPropertiesBuilder receiveMaximum(int value) {
this.receiveMaximum = checkPositive(value, "value");
return this;
}
public ConnAckPropertiesBuilder maximumQos(byte value) {
if (value != 0 && value != 1) {
throw new IllegalArgumentException("maximum QoS property could be 0 or 1");
}
this.maximumQos = value;
return this;
}
public ConnAckPropertiesBuilder retainAvailable(boolean retain) {
this.retain = retain;
return this;
}
public ConnAckPropertiesBuilder maximumPacketSize(long size) {
this.maximumPacketSize = checkPositive(size, "size");
return this;
}
public ConnAckPropertiesBuilder assignedClientId(String clientId) {
this.clientId = clientId;
return this;
}
public ConnAckPropertiesBuilder topicAliasMaximum(int value) {
this.topicAliasMaximum = value;
return this;
}
public ConnAckPropertiesBuilder reasonString(String reason) {
this.reasonString = reason;
return this;
}
public ConnAckPropertiesBuilder userProperty(String name, String value) {
userProperties.add(name, value);
return this;
}
public ConnAckPropertiesBuilder wildcardSubscriptionAvailable(boolean value) {
this.wildcardSubscriptionAvailable = value;
return this;
}
public ConnAckPropertiesBuilder subscriptionIdentifiersAvailable(boolean value) {
this.subscriptionIdentifiersAvailable = value;
return this;
}
public ConnAckPropertiesBuilder sharedSubscriptionAvailable(boolean value) {
this.sharedSubscriptionAvailable = value;
return this;
}
public ConnAckPropertiesBuilder serverKeepAlive(int seconds) {
this.serverKeepAlive = seconds;
return this;
}
public ConnAckPropertiesBuilder responseInformation(String value) {
this.responseInformation = value;
return this;
}
public ConnAckPropertiesBuilder serverReference(String host) {
this.serverReference = host;
return this;
}
public ConnAckPropertiesBuilder authenticationMethod(String methodName) {
this.authenticationMethod = methodName;
return this;
}
public ConnAckPropertiesBuilder authenticationData(byte[] rawData) {
this.authenticationData = rawData.clone();
return this;
}
}
public static final class PubAckBuilder {
private int packetId;
private byte reasonCode;
private MqttProperties properties;
PubAckBuilder() {
}
public PubAckBuilder reasonCode(byte reasonCode) {
this.reasonCode = reasonCode;
return this;
}
public PubAckBuilder packetId(int packetId) {
this.packetId = packetId;
return this;
}
/**
* @deprecated use {@link PubAckBuilder#packetId(int)} instead
*/
@Deprecated
public PubAckBuilder packetId(short packetId) {
return packetId(packetId & 0xFFFF);
}
public PubAckBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public MqttMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttPubReplyMessageVariableHeader mqttPubAckVariableHeader =
new MqttPubReplyMessageVariableHeader(packetId, reasonCode, properties);
return new MqttMessage(mqttFixedHeader, mqttPubAckVariableHeader);
}
}
public static final class SubAckBuilder {
private int packetId;
private MqttProperties properties;
private final List<MqttQoS> grantedQoses = new ArrayList<MqttQoS>();
SubAckBuilder() {
}
public SubAckBuilder packetId(int packetId) {
this.packetId = packetId;
return this;
}
/**
* @deprecated use {@link SubAckBuilder#packetId(int)} instead
*/
@Deprecated
public SubAckBuilder packetId(short packetId) {
return packetId(packetId & 0xFFFF);
}
public SubAckBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public SubAckBuilder addGrantedQos(MqttQoS qos) {
this.grantedQoses.add(qos);
return this;
}
public SubAckBuilder addGrantedQoses(MqttQoS... qoses) {
this.grantedQoses.addAll(Arrays.asList(qoses));
return this;
}
public MqttSubAckMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttMessageIdAndPropertiesVariableHeader mqttSubAckVariableHeader =
new MqttMessageIdAndPropertiesVariableHeader(packetId, properties);
//transform to primitive types
int[] grantedQoses = new int[this.grantedQoses.size()];
int i = 0;
for (MqttQoS grantedQos : this.grantedQoses) {
grantedQoses[i++] = grantedQos.value();
}
MqttSubAckPayload subAckPayload = new MqttSubAckPayload(grantedQoses);
return new MqttSubAckMessage(mqttFixedHeader, mqttSubAckVariableHeader, subAckPayload);
}
}
public static final class UnsubAckBuilder {
private int packetId;
private MqttProperties properties;
private final List<Short> reasonCodes = new ArrayList<Short>();
UnsubAckBuilder() {
}
public UnsubAckBuilder packetId(int packetId) {
this.packetId = packetId;
return this;
}
/**
* @deprecated use {@link UnsubAckBuilder#packetId(int)} instead
*/
@Deprecated
public UnsubAckBuilder packetId(short packetId) {
return packetId(packetId & 0xFFFF);
}
public UnsubAckBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public UnsubAckBuilder addReasonCode(short reasonCode) {
this.reasonCodes.add(reasonCode);
return this;
}
public UnsubAckBuilder addReasonCodes(Short... reasonCodes) {
this.reasonCodes.addAll(Arrays.asList(reasonCodes));
return this;
}
public MqttUnsubAckMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttMessageIdAndPropertiesVariableHeader mqttSubAckVariableHeader =
new MqttMessageIdAndPropertiesVariableHeader(packetId, properties);
MqttUnsubAckPayload subAckPayload = new MqttUnsubAckPayload(reasonCodes);
return new MqttUnsubAckMessage(mqttFixedHeader, mqttSubAckVariableHeader, subAckPayload);
}
}
public static final class DisconnectBuilder {
private MqttProperties properties;
private byte reasonCode;
DisconnectBuilder() {
}
public DisconnectBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public DisconnectBuilder reasonCode(byte reasonCode) {
this.reasonCode = reasonCode;
return this;
}
public MqttMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttReasonCodeAndPropertiesVariableHeader mqttDisconnectVariableHeader =
new MqttReasonCodeAndPropertiesVariableHeader(reasonCode, properties);
return new MqttMessage(mqttFixedHeader, mqttDisconnectVariableHeader);
}
}
public static final class AuthBuilder {
private MqttProperties properties;
private byte reasonCode;
AuthBuilder() {
}
public AuthBuilder properties(MqttProperties properties) {
this.properties = properties;
return this;
}
public AuthBuilder reasonCode(byte reasonCode) {
this.reasonCode = reasonCode;
return this;
}
public MqttMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.AUTH, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttReasonCodeAndPropertiesVariableHeader mqttAuthVariableHeader =
new MqttReasonCodeAndPropertiesVariableHeader(reasonCode, properties);
return new MqttMessage(mqttFixedHeader, mqttAuthVariableHeader);
}
}
public static ConnectBuilder connect() {
return new ConnectBuilder();
}
@@ -278,6 +737,26 @@ public final class MqttMessageBuilders {
return new UnsubscribeBuilder();
}
public static PubAckBuilder pubAck() {
return new PubAckBuilder();
}
public static SubAckBuilder subAck() {
return new SubAckBuilder();
}
public static UnsubAckBuilder unsubAck() {
return new UnsubAckBuilder();
}
public static DisconnectBuilder disconnect() {
return new DisconnectBuilder();
}
public static AuthBuilder auth() {
return new AuthBuilder();
}
private MqttMessageBuilders() {
}
}
+16 -3
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -50,7 +50,8 @@ public final class MqttMessageFactory {
case UNSUBACK:
return new MqttUnsubAckMessage(
mqttFixedHeader,
(MqttMessageIdVariableHeader) variableHeader);
(MqttMessageIdVariableHeader) variableHeader,
(MqttUnsubAckPayload) payload);
case UNSUBSCRIBE:
return new MqttUnsubscribeMessage(
@@ -65,17 +66,24 @@ public final class MqttMessageFactory {
(ByteBuf) payload);
case PUBACK:
//Having MqttPubReplyMessageVariableHeader or MqttMessageIdVariableHeader
return new MqttPubAckMessage(mqttFixedHeader, (MqttMessageIdVariableHeader) variableHeader);
case PUBREC:
case PUBREL:
case PUBCOMP:
//Having MqttPubReplyMessageVariableHeader or MqttMessageIdVariableHeader
return new MqttMessage(mqttFixedHeader, variableHeader);
case PINGREQ:
case PINGRESP:
case DISCONNECT:
return new MqttMessage(mqttFixedHeader);
case DISCONNECT:
case AUTH:
//Having MqttReasonCodeAndPropertiesVariableHeader
return new MqttMessage(mqttFixedHeader,
variableHeader);
default:
throw new IllegalArgumentException("unknown message type: " + mqttFixedHeader.messageType());
}
@@ -85,5 +93,10 @@ public final class MqttMessageFactory {
return new MqttMessage(null, null, null, DecoderResult.failure(cause));
}
public static MqttMessage newInvalidMessage(MqttFixedHeader mqttFixedHeader, Object variableHeader,
Throwable cause) {
return new MqttMessage(mqttFixedHeader, variableHeader, null, DecoderResult.failure(cause));
}
private MqttMessageFactory() { }
}
@@ -0,0 +1,52 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.handler.codec.mqtt;
import io.netty.util.internal.StringUtil;
/**
* Variable Header containing, Packet Id and Properties as in MQTT v5 spec.
*/
public final class MqttMessageIdAndPropertiesVariableHeader extends MqttMessageIdVariableHeader {
private final MqttProperties properties;
public MqttMessageIdAndPropertiesVariableHeader(int messageId, MqttProperties properties) {
super(messageId);
if (messageId < 1 || messageId > 0xffff) {
throw new IllegalArgumentException("messageId: " + messageId + " (expected: 1 ~ 65535)");
}
this.properties = MqttProperties.withEmptyDefaults(properties);
}
public MqttProperties properties() {
return properties;
}
@Override
public String toString() {
return StringUtil.simpleClassName(this) + "[" +
"messageId=" + messageId() +
", properties=" + properties +
']';
}
@Override
MqttMessageIdAndPropertiesVariableHeader withDefaultEmptyProperties() {
return this;
}
}
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -20,9 +20,9 @@ import io.netty.util.internal.StringUtil;
/**
* Variable Header containing only Message Id
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#msg-id">MQTTV3.1/msg-id</a>
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#msg-id">MQTTV3.1/msg-id</a>
*/
public final class MqttMessageIdVariableHeader {
public class MqttMessageIdVariableHeader {
private final int messageId;
@@ -33,7 +33,7 @@ public final class MqttMessageIdVariableHeader {
return new MqttMessageIdVariableHeader(messageId);
}
private MqttMessageIdVariableHeader(int messageId) {
protected MqttMessageIdVariableHeader(int messageId) {
this.messageId = messageId;
}
@@ -49,4 +49,12 @@ public final class MqttMessageIdVariableHeader {
.append(']')
.toString();
}
public MqttMessageIdAndPropertiesVariableHeader withEmptyProperties() {
return new MqttMessageIdAndPropertiesVariableHeader(messageId, MqttProperties.NO_PROPERTIES);
}
MqttMessageIdAndPropertiesVariableHeader withDefaultEmptyProperties() {
return withEmptyProperties();
}
}
+22 -7
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -33,7 +33,24 @@ public enum MqttMessageType {
UNSUBACK(11),
PINGREQ(12),
PINGRESP(13),
DISCONNECT(14);
DISCONNECT(14),
AUTH(15);
private static final MqttMessageType[] VALUES;
static {
// this prevent values to be assigned with the wrong order
// and ensure valueOf to work fine
final MqttMessageType[] values = values();
VALUES = new MqttMessageType[values.length + 1];
for (MqttMessageType mqttMessageType : values) {
final int value = mqttMessageType.value;
if (VALUES[value] != null) {
throw new AssertionError("value already in use: " + value);
}
VALUES[value] = mqttMessageType;
}
}
private final int value;
@@ -46,12 +63,10 @@ public enum MqttMessageType {
}
public static MqttMessageType valueOf(int type) {
for (MqttMessageType t : values()) {
if (t.value == type) {
return t;
}
if (type <= 0 || type >= VALUES.length) {
throw new IllegalArgumentException("unknown message type: " + type);
}
throw new IllegalArgumentException("unknown message type: " + type);
return VALUES[type];
}
}
@@ -0,0 +1,414 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.handler.codec.mqtt;
import io.netty.util.collection.IntObjectHashMap;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
/**
* MQTT Properties container
* */
public final class MqttProperties {
public enum MqttPropertyType {
// single byte properties
PAYLOAD_FORMAT_INDICATOR(0x01),
REQUEST_PROBLEM_INFORMATION(0x17),
REQUEST_RESPONSE_INFORMATION(0x19),
MAXIMUM_QOS(0x24),
RETAIN_AVAILABLE(0x25),
WILDCARD_SUBSCRIPTION_AVAILABLE(0x28),
SUBSCRIPTION_IDENTIFIER_AVAILABLE(0x29),
SHARED_SUBSCRIPTION_AVAILABLE(0x2A),
// two bytes properties
SERVER_KEEP_ALIVE(0x13),
RECEIVE_MAXIMUM(0x21),
TOPIC_ALIAS_MAXIMUM(0x22),
TOPIC_ALIAS(0x23),
// four bytes properties
PUBLICATION_EXPIRY_INTERVAL(0x02),
SESSION_EXPIRY_INTERVAL(0x11),
WILL_DELAY_INTERVAL(0x18),
MAXIMUM_PACKET_SIZE(0x27),
// Variable Byte Integer
SUBSCRIPTION_IDENTIFIER(0x0B),
// UTF-8 Encoded String properties
CONTENT_TYPE(0x03),
RESPONSE_TOPIC(0x08),
ASSIGNED_CLIENT_IDENTIFIER(0x12),
AUTHENTICATION_METHOD(0x15),
RESPONSE_INFORMATION(0x1A),
SERVER_REFERENCE(0x1C),
REASON_STRING(0x1F),
USER_PROPERTY(0x26),
// Binary Data
CORRELATION_DATA(0x09),
AUTHENTICATION_DATA(0x16);
private static final MqttPropertyType[] VALUES;
static {
VALUES = new MqttPropertyType[43];
for (MqttPropertyType v : values()) {
VALUES[v.value] = v;
}
}
private final int value;
MqttPropertyType(int value) {
this.value = value;
}
public int value() {
return value;
}
public static MqttPropertyType valueOf(int type) {
MqttPropertyType t = null;
try {
t = VALUES[type];
} catch (ArrayIndexOutOfBoundsException ignored) {
// nop
}
if (t == null) {
throw new IllegalArgumentException("unknown property type: " + type);
}
return t;
}
}
public static final MqttProperties NO_PROPERTIES = new MqttProperties(false);
static MqttProperties withEmptyDefaults(MqttProperties properties) {
if (properties == null) {
return MqttProperties.NO_PROPERTIES;
}
return properties;
}
/**
* MQTT property base class
*
* @param <T> property type
*/
public abstract static class MqttProperty<T> {
final T value;
final int propertyId;
protected MqttProperty(int propertyId, T value) {
this.propertyId = propertyId;
this.value = value;
}
/**
* Get MQTT property value
*
* @return property value
*/
public T value() {
return value;
}
/**
* Get MQTT property ID
* @return property ID
*/
public int propertyId() {
return propertyId;
}
@Override
public int hashCode() {
return propertyId + 31 * value.hashCode();
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj == null || getClass() != obj.getClass()) {
return false;
}
MqttProperty that = (MqttProperty) obj;
return this.propertyId == that.propertyId && this.value.equals(that.value);
}
}
public static final class IntegerProperty extends MqttProperty<Integer> {
public IntegerProperty(int propertyId, Integer value) {
super(propertyId, value);
}
@Override
public String toString() {
return "IntegerProperty(" + propertyId + ", " + value + ")";
}
}
public static final class StringProperty extends MqttProperty<String> {
public StringProperty(int propertyId, String value) {
super(propertyId, value);
}
@Override
public String toString() {
return "StringProperty(" + propertyId + ", " + value + ")";
}
}
public static final class StringPair {
public final String key;
public final String value;
public StringPair(String key, String value) {
this.key = key;
this.value = value;
}
@Override
public int hashCode() {
return key.hashCode() + 31 * value.hashCode();
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj == null || getClass() != obj.getClass()) {
return false;
}
StringPair that = (StringPair) obj;
return that.key.equals(this.key) && that.value.equals(this.value);
}
}
//User properties are the only properties that may be included multiple times and
//are the only properties where ordering is required. Therefore, they need a special handling
public static final class UserProperties extends MqttProperty<List<StringPair>> {
public UserProperties() {
super(MqttPropertyType.USER_PROPERTY.value, new ArrayList<StringPair>());
}
/**
* Create user properties from the collection of the String pair values
*
* @param values string pairs. Collection entries are copied, collection itself isn't shared
*/
public UserProperties(Collection<StringPair> values) {
this();
this.value.addAll(values);
}
private static UserProperties fromUserPropertyCollection(Collection<UserProperty> properties) {
UserProperties userProperties = new UserProperties();
for (UserProperty property: properties) {
userProperties.add(new StringPair(property.value.key, property.value.value));
}
return userProperties;
}
public void add(StringPair pair) {
this.value.add(pair);
}
public void add(String key, String value) {
this.value.add(new StringPair(key, value));
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder("UserProperties(");
boolean first = true;
for (StringPair pair: value) {
if (!first) {
builder.append(", ");
}
builder.append(pair.key + "->" + pair.value);
first = false;
}
builder.append(")");
return builder.toString();
}
}
public static final class UserProperty extends MqttProperty<StringPair> {
public UserProperty(String key, String value) {
super(MqttPropertyType.USER_PROPERTY.value, new StringPair(key, value));
}
@Override
public String toString() {
return "UserProperty(" + value.key + ", " + value.value + ")";
}
}
public static final class BinaryProperty extends MqttProperty<byte[]> {
public BinaryProperty(int propertyId, byte[] value) {
super(propertyId, value);
}
@Override
public String toString() {
return "BinaryProperty(" + propertyId + ", " + value.length + " bytes)";
}
}
public MqttProperties() {
this(true);
}
private MqttProperties(boolean canModify) {
this.canModify = canModify;
}
private IntObjectHashMap<MqttProperty> props;
private List<UserProperty> userProperties;
private List<IntegerProperty> subscriptionIds;
private final boolean canModify;
public void add(MqttProperty property) {
if (!canModify) {
throw new UnsupportedOperationException("adding property isn't allowed");
}
IntObjectHashMap<MqttProperty> props = this.props;
if (property.propertyId == MqttPropertyType.USER_PROPERTY.value) {
List<UserProperty> userProperties = this.userProperties;
if (userProperties == null) {
userProperties = new ArrayList<UserProperty>(1);
this.userProperties = userProperties;
}
if (property instanceof UserProperty) {
userProperties.add((UserProperty) property);
} else if (property instanceof UserProperties) {
for (StringPair pair: ((UserProperties) property).value) {
userProperties.add(new UserProperty(pair.key, pair.value));
}
} else {
throw new IllegalArgumentException("User property must be of UserProperty or UserProperties type");
}
} else if (property.propertyId == MqttPropertyType.SUBSCRIPTION_IDENTIFIER.value) {
List<IntegerProperty> subscriptionIds = this.subscriptionIds;
if (subscriptionIds == null) {
subscriptionIds = new ArrayList<IntegerProperty>(1);
this.subscriptionIds = subscriptionIds;
}
if (property instanceof IntegerProperty) {
subscriptionIds.add((IntegerProperty) property);
} else {
throw new IllegalArgumentException("Subscription ID must be an integer property");
}
} else {
if (props == null) {
props = new IntObjectHashMap<MqttProperty>();
this.props = props;
}
props.put(property.propertyId, property);
}
}
public Collection<? extends MqttProperty> listAll() {
IntObjectHashMap<MqttProperty> props = this.props;
if (props == null && subscriptionIds == null && userProperties == null) {
return Collections.<MqttProperty>emptyList();
}
if (subscriptionIds == null && userProperties == null) {
return props.values();
}
if (props == null && userProperties == null) {
return subscriptionIds;
}
List<MqttProperty> propValues = new ArrayList<MqttProperty>(props != null ? props.size() : 1);
if (props != null) {
propValues.addAll(props.values());
}
if (subscriptionIds != null) {
propValues.addAll(subscriptionIds);
}
if (userProperties != null) {
propValues.add(UserProperties.fromUserPropertyCollection(userProperties));
}
return propValues;
}
public boolean isEmpty() {
IntObjectHashMap<MqttProperty> props = this.props;
return props == null || props.isEmpty();
}
/**
* Get property by ID. If there are multiple properties of this type (can be with Subscription ID)
* then return the first one.
*
* @param propertyId ID of the property
* @return a property if it is set, null otherwise
*/
public MqttProperty getProperty(int propertyId) {
if (propertyId == MqttPropertyType.USER_PROPERTY.value) {
//special handling to keep compatibility with earlier versions
List<UserProperty> userProperties = this.userProperties;
if (userProperties == null) {
return null;
}
return UserProperties.fromUserPropertyCollection(userProperties);
}
if (propertyId == MqttPropertyType.SUBSCRIPTION_IDENTIFIER.value) {
List<IntegerProperty> subscriptionIds = this.subscriptionIds;
if (subscriptionIds == null || subscriptionIds.isEmpty()) {
return null;
}
return subscriptionIds.get(0);
}
IntObjectHashMap<MqttProperty> props = this.props;
return props == null ? null : props.get(propertyId);
}
/**
* Get properties by ID.
* Some properties (Subscription ID and User Properties) may occur multiple times,
* this method returns all their values in order.
*
* @param propertyId ID of the property
* @return all properties having specified ID
*/
public List<? extends MqttProperty> getProperties(int propertyId) {
if (propertyId == MqttPropertyType.USER_PROPERTY.value) {
return userProperties == null ? Collections.<MqttProperty>emptyList() : userProperties;
}
if (propertyId == MqttPropertyType.SUBSCRIPTION_IDENTIFIER.value) {
return subscriptionIds == null ? Collections.<MqttProperty>emptyList() : subscriptionIds;
}
IntObjectHashMap<MqttProperty> props = this.props;
return (props == null || !props.containsKey(propertyId)) ?
Collections.<MqttProperty>emptyList() :
Collections.singletonList(props.get(propertyId));
}
}
+2 -2
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -17,7 +17,7 @@
package io.netty.handler.codec.mqtt;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#puback">MQTTV3.1/puback</a>
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#puback">MQTTV3.1/puback</a>
*/
public final class MqttPubAckMessage extends MqttMessage {
@@ -0,0 +1,56 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.handler.codec.mqtt;
import io.netty.util.internal.StringUtil;
/**
* Variable Header containing Packet Id, reason code and Properties as in MQTT v5 spec.
*/
public final class MqttPubReplyMessageVariableHeader extends MqttMessageIdVariableHeader {
private final byte reasonCode;
private final MqttProperties properties;
public static final byte REASON_CODE_OK = 0;
public MqttPubReplyMessageVariableHeader(int messageId, byte reasonCode, MqttProperties properties) {
super(messageId);
if (messageId < 1 || messageId > 0xffff) {
throw new IllegalArgumentException("messageId: " + messageId + " (expected: 1 ~ 65535)");
}
this.reasonCode = reasonCode;
this.properties = MqttProperties.withEmptyDefaults(properties);
}
public byte reasonCode() {
return reasonCode;
}
public MqttProperties properties() {
return properties;
}
@Override
public String toString() {
return StringUtil.simpleClassName(this) + "[" +
"messageId=" + messageId() +
", reasonCode=" + reasonCode +
", properties=" + properties +
']';
}
}
+4 -8
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -18,10 +18,10 @@ package io.netty.handler.codec.mqtt;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.util.IllegalReferenceCountException;
import io.netty.buffer.ByteBufUtil;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#publish">MQTTV3.1/publish</a>
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#publish">MQTTV3.1/publish</a>
*/
public class MqttPublishMessage extends MqttMessage implements ByteBufHolder {
@@ -44,11 +44,7 @@ public class MqttPublishMessage extends MqttMessage implements ByteBufHolder {
@Override
public ByteBuf content() {
final ByteBuf data = (ByteBuf) super.payload();
if (data.refCnt() <= 0) {
throw new IllegalReferenceCountException(data.refCnt());
}
return data;
return ByteBufUtil.ensureAccessible((ByteBuf) super.payload());
}
@Override
+16 -2
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -24,11 +24,17 @@ import io.netty.util.internal.StringUtil;
public final class MqttPublishVariableHeader {
private final String topicName;
private final int packetId;
private int packetId;
private final MqttProperties properties;
public MqttPublishVariableHeader(String topicName, int packetId) {
this(topicName, packetId, MqttProperties.NO_PROPERTIES);
}
public MqttPublishVariableHeader(String topicName, int packetId, MqttProperties properties) {
this.topicName = topicName;
this.packetId = packetId;
this.properties = MqttProperties.withEmptyDefaults(properties);
}
public String topicName() {
@@ -47,6 +53,14 @@ public final class MqttPublishVariableHeader {
return packetId;
}
public MqttProperties properties() {
return properties;
}
public void setPacketId(int packetId) {
this.packetId = packetId;
}
@Override
public String toString() {
return new StringBuilder(StringUtil.simpleClassName(this))
+12 -6
View File
@@ -5,7 +5,7 @@
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
@@ -32,11 +32,17 @@ public enum MqttQoS {
}
public static MqttQoS valueOf(int value) {
for (MqttQoS q: values()) {
if (q.value == value) {
return q;
}
switch (value) {
case 0:
return AT_MOST_ONCE;
case 1:
return AT_LEAST_ONCE;
case 2:
return EXACTLY_ONCE;
case 0x80:
return FAILURE;
default:
throw new IllegalArgumentException("invalid QoS: " + value);
}
throw new IllegalArgumentException("invalid QoS: " + value);
}
}
@@ -0,0 +1,54 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.handler.codec.mqtt;
import io.netty.util.internal.StringUtil;
/**
* Variable Header for AUTH and DISCONNECT messages represented by {@link MqttMessage}
*/
public final class MqttReasonCodeAndPropertiesVariableHeader {
private final byte reasonCode;
private final MqttProperties properties;
public static final byte REASON_CODE_OK = 0;
public MqttReasonCodeAndPropertiesVariableHeader(byte reasonCode,
MqttProperties properties) {
this.reasonCode = reasonCode;
this.properties = MqttProperties.withEmptyDefaults(properties);
}
public byte reasonCode() {
return reasonCode;
}
public MqttProperties properties() {
return properties;
}
@Override
public String toString() {
return new StringBuilder(StringUtil.simpleClassName(this))
.append('[')
.append("reasonCode=").append(reasonCode)
.append(", properties=").append(properties)
.append(']')
.toString();
}
}
+14 -3
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -17,22 +17,33 @@
package io.netty.handler.codec.mqtt;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#suback">MQTTV3.1/suback</a>
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#suback">MQTTV3.1/suback</a>
*/
public final class MqttSubAckMessage extends MqttMessage {
public MqttSubAckMessage(
MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttMessageIdAndPropertiesVariableHeader variableHeader,
MqttSubAckPayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
public MqttSubAckMessage(
MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttSubAckPayload payload) {
this(mqttFixedHeader, variableHeader.withDefaultEmptyProperties(), payload);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
public MqttMessageIdAndPropertiesVariableHeader idAndPropertiesVariableHeader() {
return (MqttMessageIdAndPropertiesVariableHeader) super.variableHeader();
}
@Override
public MqttSubAckPayload payload() {
return (MqttSubAckPayload) super.payload();
+26 -17
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -16,6 +16,7 @@
package io.netty.handler.codec.mqtt;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.StringUtil;
import java.util.ArrayList;
@@ -27,43 +28,51 @@ import java.util.List;
*/
public class MqttSubAckPayload {
private final List<Integer> grantedQoSLevels;
private final List<Integer> reasonCodes;
public MqttSubAckPayload(int... grantedQoSLevels) {
if (grantedQoSLevels == null) {
throw new NullPointerException("grantedQoSLevels");
}
public MqttSubAckPayload(int... reasonCodes) {
ObjectUtil.checkNotNull(reasonCodes, "reasonCodes");
List<Integer> list = new ArrayList<Integer>(grantedQoSLevels.length);
for (int v: grantedQoSLevels) {
List<Integer> list = new ArrayList<Integer>(reasonCodes.length);
for (int v: reasonCodes) {
list.add(v);
}
this.grantedQoSLevels = Collections.unmodifiableList(list);
this.reasonCodes = Collections.unmodifiableList(list);
}
public MqttSubAckPayload(Iterable<Integer> grantedQoSLevels) {
if (grantedQoSLevels == null) {
throw new NullPointerException("grantedQoSLevels");
}
public MqttSubAckPayload(Iterable<Integer> reasonCodes) {
ObjectUtil.checkNotNull(reasonCodes, "reasonCodes");
List<Integer> list = new ArrayList<Integer>();
for (Integer v: grantedQoSLevels) {
for (Integer v: reasonCodes) {
if (v == null) {
break;
}
list.add(v);
}
this.grantedQoSLevels = Collections.unmodifiableList(list);
this.reasonCodes = Collections.unmodifiableList(list);
}
public List<Integer> grantedQoSLevels() {
return grantedQoSLevels;
List<Integer> qosLevels = new ArrayList<Integer>(reasonCodes.size());
for (int code: reasonCodes) {
if (code > MqttQoS.EXACTLY_ONCE.value()) {
qosLevels.add(MqttQoS.FAILURE.value());
} else {
qosLevels.add(code);
}
}
return qosLevels;
}
public List<Integer> reasonCodes() {
return reasonCodes;
}
@Override
public String toString() {
return new StringBuilder(StringUtil.simpleClassName(this))
.append('[')
.append("grantedQoSLevels=").append(grantedQoSLevels)
.append("reasonCodes=").append(reasonCodes)
.append(']')
.toString();
}
+14 -3
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -17,23 +17,34 @@
package io.netty.handler.codec.mqtt;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#subscribe">
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#subscribe">
* MQTTV3.1/subscribe</a>
*/
public final class MqttSubscribeMessage extends MqttMessage {
public MqttSubscribeMessage(
MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttMessageIdAndPropertiesVariableHeader variableHeader,
MqttSubscribePayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
public MqttSubscribeMessage(
MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttSubscribePayload payload) {
this(mqttFixedHeader, variableHeader.withDefaultEmptyProperties(), payload);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
public MqttMessageIdAndPropertiesVariableHeader idAndPropertiesVariableHeader() {
return (MqttMessageIdAndPropertiesVariableHeader) super.variableHeader();
}
@Override
public MqttSubscribePayload payload() {
return (MqttSubscribePayload) super.payload();
+6 -5
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -39,11 +39,12 @@ public final class MqttSubscribePayload {
@Override
public String toString() {
StringBuilder builder = new StringBuilder(StringUtil.simpleClassName(this)).append('[');
for (int i = 0; i < topicSubscriptions.size() - 1; i++) {
for (int i = 0; i < topicSubscriptions.size(); i++) {
builder.append(topicSubscriptions.get(i)).append(", ");
}
builder.append(topicSubscriptions.get(topicSubscriptions.size() - 1));
builder.append(']');
return builder.toString();
if (!topicSubscriptions.isEmpty()) {
builder.setLength(builder.length() - 2);
}
return builder.append(']').toString();
}
}
@@ -0,0 +1,128 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.handler.codec.mqtt;
/**
* Model the SubscriptionOption used in Subscribe MQTT v5 packet
*/
public final class MqttSubscriptionOption {
public enum RetainedHandlingPolicy {
SEND_AT_SUBSCRIBE(0),
SEND_AT_SUBSCRIBE_IF_NOT_YET_EXISTS(1),
DONT_SEND_AT_SUBSCRIBE(2);
private final int value;
RetainedHandlingPolicy(int value) {
this.value = value;
}
public int value() {
return value;
}
public static RetainedHandlingPolicy valueOf(int value) {
switch (value) {
case 0:
return SEND_AT_SUBSCRIBE;
case 1:
return SEND_AT_SUBSCRIBE_IF_NOT_YET_EXISTS;
case 2:
return DONT_SEND_AT_SUBSCRIBE;
default:
throw new IllegalArgumentException("invalid RetainedHandlingPolicy: " + value);
}
}
}
private final MqttQoS qos;
private final boolean noLocal;
private final boolean retainAsPublished;
private final RetainedHandlingPolicy retainHandling;
public static MqttSubscriptionOption onlyFromQos(MqttQoS qos) {
return new MqttSubscriptionOption(qos, false, false, RetainedHandlingPolicy.SEND_AT_SUBSCRIBE);
}
public MqttSubscriptionOption(MqttQoS qos,
boolean noLocal,
boolean retainAsPublished,
RetainedHandlingPolicy retainHandling) {
this.qos = qos;
this.noLocal = noLocal;
this.retainAsPublished = retainAsPublished;
this.retainHandling = retainHandling;
}
public MqttQoS qos() {
return qos;
}
public boolean isNoLocal() {
return noLocal;
}
public boolean isRetainAsPublished() {
return retainAsPublished;
}
public RetainedHandlingPolicy retainHandling() {
return retainHandling;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
MqttSubscriptionOption that = (MqttSubscriptionOption) o;
if (noLocal != that.noLocal) {
return false;
}
if (retainAsPublished != that.retainAsPublished) {
return false;
}
if (qos != that.qos) {
return false;
}
return retainHandling == that.retainHandling;
}
@Override
public int hashCode() {
int result = qos.hashCode();
result = 31 * result + (noLocal ? 1 : 0);
result = 31 * result + (retainAsPublished ? 1 : 0);
result = 31 * result + retainHandling.hashCode();
return result;
}
@Override
public String toString() {
return "SubscriptionOption[" +
"qos=" + qos +
", noLocal=" + noLocal +
", retainAsPublished=" + retainAsPublished +
", retainHandling=" + retainHandling +
']';
}
}
+14 -5
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -25,11 +25,16 @@ import io.netty.util.internal.StringUtil;
public final class MqttTopicSubscription {
private final String topicFilter;
private final MqttQoS qualityOfService;
private final MqttSubscriptionOption option;
public MqttTopicSubscription(String topicFilter, MqttQoS qualityOfService) {
this.topicFilter = topicFilter;
this.qualityOfService = qualityOfService;
this.option = MqttSubscriptionOption.onlyFromQos(qualityOfService);
}
public MqttTopicSubscription(String topicFilter, MqttSubscriptionOption option) {
this.topicFilter = topicFilter;
this.option = option;
}
public String topicName() {
@@ -37,7 +42,11 @@ public final class MqttTopicSubscription {
}
public MqttQoS qualityOfService() {
return qualityOfService;
return option.qos();
}
public MqttSubscriptionOption option() {
return option;
}
@Override
@@ -45,7 +54,7 @@ public final class MqttTopicSubscription {
return new StringBuilder(StringUtil.simpleClassName(this))
.append('[')
.append("topicFilter=").append(topicFilter)
.append(", qualityOfService=").append(qualityOfService)
.append(", option=").append(this.option)
.append(']')
.toString();
}
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+34 -4
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -17,16 +17,46 @@
package io.netty.handler.codec.mqtt;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#unsuback">MQTTV3.1/unsuback</a>
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#unsuback">
* MQTTV3.1/unsuback</a>
*/
public final class MqttUnsubAckMessage extends MqttMessage {
public MqttUnsubAckMessage(MqttFixedHeader mqttFixedHeader, MqttMessageIdVariableHeader variableHeader) {
super(mqttFixedHeader, variableHeader, null);
public MqttUnsubAckMessage(MqttFixedHeader mqttFixedHeader,
MqttMessageIdAndPropertiesVariableHeader variableHeader,
MqttUnsubAckPayload payload) {
super(mqttFixedHeader, variableHeader, MqttUnsubAckPayload.withEmptyDefaults(payload));
}
public MqttUnsubAckMessage(MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttUnsubAckPayload payload) {
this(mqttFixedHeader, fallbackVariableHeader(variableHeader), payload);
}
public MqttUnsubAckMessage(MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader) {
this(mqttFixedHeader, variableHeader, null);
}
private static MqttMessageIdAndPropertiesVariableHeader fallbackVariableHeader(
MqttMessageIdVariableHeader variableHeader) {
if (variableHeader instanceof MqttMessageIdAndPropertiesVariableHeader) {
return (MqttMessageIdAndPropertiesVariableHeader) variableHeader;
}
return new MqttMessageIdAndPropertiesVariableHeader(variableHeader.messageId(), MqttProperties.NO_PROPERTIES);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
public MqttMessageIdAndPropertiesVariableHeader idAndPropertiesVariableHeader() {
return (MqttMessageIdAndPropertiesVariableHeader) super.variableHeader();
}
@Override
public MqttUnsubAckPayload payload() {
return (MqttUnsubAckPayload) super.payload();
}
}
@@ -0,0 +1,75 @@
/*
* Copyright 2020 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.handler.codec.mqtt;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.StringUtil;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
/**
* Payload for MQTT unsuback message as in V5.
*/
public final class MqttUnsubAckPayload {
private final List<Short> unsubscribeReasonCodes;
private static final MqttUnsubAckPayload EMPTY = new MqttUnsubAckPayload();
public static MqttUnsubAckPayload withEmptyDefaults(MqttUnsubAckPayload payload) {
if (payload == null) {
return EMPTY;
} else {
return payload;
}
}
public MqttUnsubAckPayload(short... unsubscribeReasonCodes) {
ObjectUtil.checkNotNull(unsubscribeReasonCodes, "unsubscribeReasonCodes");
List<Short> list = new ArrayList<Short>(unsubscribeReasonCodes.length);
for (Short v: unsubscribeReasonCodes) {
list.add(v);
}
this.unsubscribeReasonCodes = Collections.unmodifiableList(list);
}
public MqttUnsubAckPayload(Iterable<Short> unsubscribeReasonCodes) {
ObjectUtil.checkNotNull(unsubscribeReasonCodes, "unsubscribeReasonCodes");
List<Short> list = new ArrayList<Short>();
for (Short v: unsubscribeReasonCodes) {
ObjectUtil.checkNotNull(v, "unsubscribeReasonCode");
list.add(v);
}
this.unsubscribeReasonCodes = Collections.unmodifiableList(list);
}
public List<Short> unsubscribeReasonCodes() {
return unsubscribeReasonCodes;
}
@Override
public String toString() {
return new StringBuilder(StringUtil.simpleClassName(this))
.append('[')
.append("unsubscribeReasonCodes=").append(unsubscribeReasonCodes)
.append(']')
.toString();
}
}
+14 -3
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -17,23 +17,34 @@
package io.netty.handler.codec.mqtt;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#unsubscribe">
* See <a href="https://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#unsubscribe">
* MQTTV3.1/unsubscribe</a>
*/
public final class MqttUnsubscribeMessage extends MqttMessage {
public MqttUnsubscribeMessage(
MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttMessageIdAndPropertiesVariableHeader variableHeader,
MqttUnsubscribePayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
public MqttUnsubscribeMessage(
MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttUnsubscribePayload payload) {
this(mqttFixedHeader, variableHeader.withDefaultEmptyProperties(), payload);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
public MqttMessageIdAndPropertiesVariableHeader idAndPropertiesVariableHeader() {
return (MqttMessageIdAndPropertiesVariableHeader) super.variableHeader();
}
@Override
public MqttUnsubscribePayload payload() {
return (MqttUnsubscribePayload) super.payload();
+6 -5
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -39,11 +39,12 @@ public final class MqttUnsubscribePayload {
@Override
public String toString() {
StringBuilder builder = new StringBuilder(StringUtil.simpleClassName(this)).append('[');
for (int i = 0; i < topics.size() - 1; i++) {
for (int i = 0; i < topics.size(); i++) {
builder.append("topicName = ").append(topics.get(i)).append(", ");
}
builder.append("topicName = ").append(topics.get(topics.size() - 1))
.append(']');
return builder.toString();
if (!topics.isEmpty()) {
builder.setLength(builder.length() - 2);
}
return builder.append("]").toString();
}
}
+25 -15
View File
@@ -5,7 +5,7 @@
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
@@ -25,7 +25,7 @@ import io.netty.util.internal.ObjectUtil;
public enum MqttVersion {
MQTT_3_1("MQIsdp", (byte) 3),
MQTT_3_1_1("MQTT", (byte) 4),
Wildfire_1("MQTT", (byte) 5),
MQTT_5("MQTT", (byte) 5),
Wildfire_2("MQTT", (byte) 6),
Wildfire_Future("MQTT", (byte) 7);
@@ -50,21 +50,31 @@ public enum MqttVersion {
}
public static MqttVersion fromProtocolNameAndLevel(String protocolName, byte protocolLevel) {
for (MqttVersion mv : values()) {
if (mv.name.equals(protocolName)) {
if (mv.level == protocolLevel) {
return mv;
} else {
continue;
}
}
MqttVersion mv = null;
switch (protocolLevel) {
case 3:
mv = MQTT_3_1;
break;
case 4:
mv = MQTT_3_1_1;
break;
case 5:
mv = MQTT_5;
break;
case 6:
mv = Wildfire_2;
break;
case 7:
mv = Wildfire_Future;
break;
default:
mv = Wildfire_Future;
break;
}
if(protocolLevel > Wildfire_Future.level) {
return Wildfire_Future;
}
throw new MqttUnacceptableProtocolVersionException(protocolName + "is unknown protocol name");
}
return mv;
}
public static MqttVersion fromProtocolLevel(int protocolLevel) {
for (MqttVersion mv : values()) {
if (mv.level == (byte )protocolLevel) {
+1 -1
View File
@@ -5,7 +5,7 @@
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+1 -1
View File
@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<netty.version>4.1.12.Final</netty.version>
<netty.version>4.1.68.Final</netty.version>
<source.version>1.8</source.version>
<target.version>1.8</target.version>
</properties>