增加协议版本，新版本添加对gzip的支持

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<String, Session> sessions = new ConcurrentHashMap<>();

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);

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 {

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;

broker/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java

@@ -223,7 +223,7 @@ 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 == 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

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");
+    }
 }