fix: 解决遗嘱消息缓存部分丢失问题 & 发布1.0.6-netty

Author: Wizzercn

mqtt-auth/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.0.5-netty</version>
+        <version>1.0.6-netty</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

mqtt-broker/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.0.5-netty</version>
+        <version>1.0.6-netty</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

mqtt-common/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.0.5-netty</version>
+        <version>1.0.6-netty</version>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>

mqtt-common/src/main/java/cn/wizzer/iot/mqtt/server/common/session/SessionStore.java

@@ -23,6 +23,10 @@ public class SessionStore implements Serializable {
 
     private MqttPublishMessage willMessage;
 
+    public SessionStore(){
+
+    }
+
     public SessionStore(String clientId, String channelId, boolean cleanSession, MqttPublishMessage willMessage) {
         this.clientId = clientId;
         this.channelId = channelId;

mqtt-store/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>mqtt-wk</artifactId>
         <groupId>cn.wizzer</groupId>
-        <version>1.0.5-netty</version>
+        <version>1.0.6-netty</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

mqtt-store/src/main/java/cn/wizzer/iot/mqtt/server/store/session/SessionStoreService.java

@@ -1,17 +1,19 @@
 /**
- * Copyright (c) 2018, Mr.Wang ([email protected]) All rights reserved.
+ * Created by wizzer on 2018
  */
 
 package cn.wizzer.iot.mqtt.server.store.session;
 
 import cn.wizzer.iot.mqtt.server.common.session.ISessionStoreService;
 import cn.wizzer.iot.mqtt.server.common.session.SessionStore;
+import cn.wizzer.iot.mqtt.server.store.util.StoreUtil;
+import com.alibaba.fastjson.JSON;
 import org.nutz.aop.interceptor.async.Async;
 import org.nutz.integration.jedis.RedisService;
 import org.nutz.ioc.loader.annotation.Inject;
 import org.nutz.ioc.loader.annotation.IocBean;
-import org.nutz.json.Json;
-import org.nutz.json.JsonFormat;
+import org.nutz.lang.Strings;
+import org.nutz.lang.util.NutMap;
 
 /**
  * 会话存储服务
@@ -24,16 +26,19 @@ public class SessionStoreService implements ISessionStoreService {
 
     @Override
     public void put(String clientId, SessionStore sessionStore) {
-        //fastjson需要对象有get/set方法，而MqttPublishMessage对象没有get/set方法造成转换失败，改成nutz的工具类
-        redisService.set(CACHE_PRE + clientId, Json.toJson(sessionStore, JsonFormat.compact()));
+        //SessionStore对象不能正常转为JSON,使用工具类类解决
+        NutMap nutMap = StoreUtil.transPublishToMapBeta(sessionStore);
+        redisService.set(CACHE_PRE + clientId, JSON.toJSONString(nutMap));
     }
 
 
     @Override
     public SessionStore get(String clientId) {
-        String obj = redisService.get(CACHE_PRE + clientId);
-        if (obj != null)
-            return Json.fromJson(SessionStore.class, obj);
+        String jsonObj = redisService.get(CACHE_PRE + clientId);
+        if (Strings.isNotBlank(jsonObj)) {
+            NutMap nutMap = JSON.parseObject(jsonObj, NutMap.class);
+            return StoreUtil.mapTransToPublishMsgBeta(nutMap);
+        }
         return null;
     }
 

mqtt-store/src/main/java/cn/wizzer/iot/mqtt/server/store/util/StoreUtil.java

@@ -0,0 +1,71 @@
+package cn.wizzer.iot.mqtt.server.store.util;
+
+import cn.wizzer.iot.mqtt.server.common.session.SessionStore;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.handler.codec.mqtt.*;
+import org.nutz.lang.util.NutMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Created by cs on 2018
+ */
+public class StoreUtil {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(StoreUtil.class);
+
+    public static NutMap transPublishToMapBeta(SessionStore store) {
+        MqttPublishMessage msg = store.getWillMessage();
+        try {
+            NutMap sessionStore = new NutMap();
+            sessionStore.addv("clientId", store.getClientId());
+            sessionStore.addv("channelId", store.getChannelId());
+            sessionStore.addv("cleanSession", store.isCleanSession());
+            sessionStore.addv("payload", new String(msg.payload().array(), "UTF-8"));
+
+            sessionStore.addv("messageType", msg.fixedHeader().messageType().value());
+            sessionStore.addv("isDup", msg.fixedHeader().isDup());
+            sessionStore.addv("qosLevel", msg.fixedHeader().qosLevel().value());
+            sessionStore.addv("isRetain", msg.fixedHeader().isRetain());
+            sessionStore.addv("remainingLength", msg.fixedHeader().remainingLength());
+
+            sessionStore.addv("topicName", msg.variableHeader().topicName());
+            sessionStore.addv("packetId", msg.variableHeader().packetId());
+
+
+            return sessionStore;
+        } catch (Exception e) {
+            LOGGER.error(e.getMessage(), e);
+        }
+        return null;
+    }
+
+    public static SessionStore mapTransToPublishMsgBeta(NutMap store) {
+        SessionStore sessionStore = new SessionStore();
+
+        String payload = store.getString("payload");
+        ByteBuf buf = ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT, payload);
+
+        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(
+                MqttMessageType.valueOf(store.getInt("messageType")),
+                store.getBoolean("isDup"),
+                MqttQoS.valueOf(store.getInt("qosLevel")),
+                store.getBoolean("isRetain"),
+                store.getInt("remainingLength"));
+
+        MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(store.getString("topicName"),
+                store.getInt("packetId"));
+
+        MqttPublishMessage mqttPublishMessage = new MqttPublishMessage(mqttFixedHeader, mqttPublishVariableHeader, buf);
+
+
+        sessionStore.setChannelId(store.getString("channelId"));
+        sessionStore.setClientId(store.getString("clientId"));
+        sessionStore.setCleanSession(store.getBoolean("cleanSession"));
+        sessionStore.setWillMessage(mqttPublishMessage);
+        return sessionStore;
+    }
+
+}

pom.xml

@@ -6,7 +6,7 @@
     <groupId>cn.wizzer</groupId>
     <artifactId>mqtt-wk</artifactId>
     <packaging>pom</packaging>
-    <version>1.0.5-netty</version>
+    <version>1.0.6-netty</version>
     <name>MqttWk</name>
     <modules>
         <module>mqtt-common</module>
@@ -15,7 +15,7 @@
         <module>mqtt-store</module>
     </modules>
     <properties>
-        <mqttwk.version>1.0.3-netty</mqttwk.version>
+        <mqttwk.version>1.0.6-netty</mqttwk.version>
         <nutzboot.version>2.3-SNAPSHOT</nutzboot.version>
         <netty.version>4.1.28.Final</netty.version>
         <fastjson.version>1.2.49</fastjson.version>