diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index b03ad509..b9d6c256 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -23,6 +23,7 @@ import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; +import io.netty.handler.codec.mqtt.MqttVersion; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import cn.wildfirechat.common.ErrorCode; @@ -49,6 +50,8 @@ public class MemorySessionStore implements ISessionsStore { private volatile int unReceivedMsgs; + private MqttVersion mqttVersion = MqttVersion.MQTT_3_1_1; + public long getLastActiveTime() { return lastActiveTime; } @@ -230,7 +233,13 @@ public class MemorySessionStore implements ISessionsStore { } - + public MqttVersion getMqttVersion() { + return mqttVersion; + } + + public void setMqttVersion(MqttVersion mqttVersion) { + this.mqttVersion = mqttVersion; + } } private final Map sessions = new ConcurrentHashMap<>(); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index f87965ee..f809ab63 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -143,7 +143,8 @@ public class ProtocolProcessor { LOG.info("Processing CONNECT message. CId={}, username={}", clientId, payload.userName()); if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel() - && msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) { + && msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel() + && msg.variableHeader().version() != MqttVersion.Wildfire_1.protocolLevel()) { MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION); LOG.error("MQTT protocol version is not valid. CId={}", clientId); @@ -162,7 +163,8 @@ public class ProtocolProcessor { } - if (!login(channel, msg, clientId)) { + MqttVersion mqttVersion = MqttVersion.fromProtocolLevel(msg.variableHeader().version()); + if (!login(channel, msg, clientId, mqttVersion)) { channel.close(); return; } @@ -238,7 +240,7 @@ public class ProtocolProcessor { MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent); return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader, new MqttConnectAckPayload(data)); } - private boolean login(Channel channel, MqttConnectMessage msg, final String clientId) { + private boolean login(Channel channel, MqttConnectMessage msg, final String clientId, MqttVersion mqttVersion) { // handle user authentication if (msg.variableHeader().hasUserName()) { int status = m_messagesStore.getUserStatus(msg.payload().userName()); @@ -269,6 +271,7 @@ public class ProtocolProcessor { failedCredentials(channel); return false; } + session.setMqttVersion(mqttVersion); } else if (!this.allowAnonymous) { LOG.error("Client didn't supply any password and MQTT anonymous mode is disabled CId={}", clientId); failedCredentials(channel); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 413fa7d3..a1aea010 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -41,6 +41,7 @@ import io.moquette.server.ConnectionDescriptor; import io.moquette.server.Server; import io.moquette.spi.ClientSession; import io.moquette.spi.impl.security.AES; +import io.netty.handler.codec.mqtt.MqttVersion; import io.netty.util.ReferenceCountUtil; import io.netty.util.internal.StringUtil; import org.slf4j.Logger; @@ -177,7 +178,7 @@ public class Qos1PublishHandler extends QosPublishHandler { } else { MemorySessionStore.Session session = m_sessionStore.getSession(clientID); if (session != null && session.getUsername().equals(fromUser)) { - if (data.length > 10*1024) { + if (data.length > 7*1024 && session.getMqttVersion() == MqttVersion.Wildfire_1) { ByteArrayOutputStream out = new ByteArrayOutputStream(); GZIPOutputStream gzip; try { diff --git a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java index 06556795..2c31db07 100755 --- a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java +++ b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttCodecUtil.java @@ -43,7 +43,7 @@ final class MqttCodecUtil { return clientId != null && clientId.length() >= MIN_CLIENT_ID_LENGTH && clientId.length() <= MAX_CLIENT_ID_LENGTH; } - if (mqttVersion == MqttVersion.MQTT_3_1_1) { + if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.Wildfire_1) { // In 3.1.3.1 Client Identifier of MQTT 3.1.1 specification, The Server MAY allow ClientId’s // that contain more than 23 encoded bytes. And, The Server MAY allow zero-length ClientId. return clientId != null; diff --git a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java index 15d89999..ff66cdea 100755 --- a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java +++ b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java @@ -223,7 +223,7 @@ public final class MqttDecoder extends ReplayingDecoder { final int willQos = (b1 & 0x18) >> 3; final boolean willFlag = (b1 & 0x04) == 0x04; final boolean cleanSession = (b1 & 0x02) == 0x02; - if (mqttVersion == MqttVersion.MQTT_3_1_1) { + if (mqttVersion == MqttVersion.MQTT_3_1_1 || mqttVersion == MqttVersion.Wildfire_1) { 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 diff --git a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java index 985bd9a0..cd7f6519 100755 --- a/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java +++ b/broker/src/main/java/io/netty/handler/codec/mqtt/MqttVersion.java @@ -24,7 +24,8 @@ import io.netty.util.internal.ObjectUtil; */ public enum MqttVersion { MQTT_3_1("MQIsdp", (byte) 3), - MQTT_3_1_1("MQTT", (byte) 4); + MQTT_3_1_1("MQTT", (byte) 4), + Wildfire_1("MQTT", (byte) 5); private final String name; private final byte level; @@ -52,11 +53,19 @@ public enum MqttVersion { if (mv.level == protocolLevel) { return mv; } else { - throw new MqttUnacceptableProtocolVersionException(protocolName + " and " + - protocolLevel + " are not match"); + continue; } } } throw new MqttUnacceptableProtocolVersionException(protocolName + "is unknown protocol name"); } + + public static MqttVersion fromProtocolLevel(int protocolLevel) { + for (MqttVersion mv : values()) { + if (mv.level == (byte )protocolLevel) { + return mv; + } + } + throw new MqttUnacceptableProtocolVersionException(protocolLevel + "is unknown protocol level"); + } }