添加消息漫游开关

broker/src/main/java/io/moquette/BrokerConstants.java

@@ -96,7 +96,7 @@ public final class BrokerConstants {
     public static final String HTTP_SERVER_API_NO_CHECK_TIME = "http.admin.no_check_time";
 
     public static final String CLIENT_PROTO_SECRET_KEY = "client.proto.secret_key";
-    public static final String TOKEN_SECRET_KEY = "token.key ";
+    public static final String TOKEN_SECRET_KEY = "token.key";
 
     public static final String EMBED_DB_PROPERTY_NAME = "embed.db";
 
@@ -106,6 +106,7 @@ public final class BrokerConstants {
 
     public static final String MESSAGE_Forward_Url = "message.forward.url";
 
+    public static final String MESSAGE_ROAMING = "message.roaming";
 
 
     private BrokerConstants() {

broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java

@@ -82,6 +82,8 @@ public class MemoryMessagesStore implements IMessagesStore {
 
     private static final String CHANNEL_LISTENERS = "channel_listeners";
 
+    private static boolean IS_MESSAGE_ROAMING = true;
+
     static final String USER_ROBOTS = "user_robots";
     static final String USER_THINGS = "user_things";
 
@@ -115,6 +117,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     MemoryMessagesStore(Server server, DatabaseStore databaseStore) {
         m_Server = server;
         this.databaseStore = databaseStore;
+        IS_MESSAGE_ROAMING = "1".equals(m_Server.getConfig().getProperty(MESSAGE_ROAMING));
     }
 
     @Override
@@ -314,6 +317,10 @@ public class MemoryMessagesStore implements IMessagesStore {
                 });
             }
 
+            boolean noRoaming = false;
+            if (pullType == ProtoConstants.PullType.Pull_Normal && fromMessageId == 0 && !IS_MESSAGE_ROAMING) {
+                noRoaming = true;
+            }
             while (true) {
                 Map.Entry<Long, Long> entry = maps.higherEntry(current);
                 if (entry == null) {
@@ -333,6 +340,10 @@ public class MemoryMessagesStore implements IMessagesStore {
                             }
                         }
 
+                        if (noRoaming && (System.currentTimeMillis() - bundle.getMessage().getServerTimestamp() > 5 * 60 * 1000)) {
+                            continue;
+                        }
+
                         size += bundle.getMessage().getSerializedSize();
                         if (size >= 1 * 1024 * 1024) { //3M
                             break;