diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index faf95dc0..5acf3867 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/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() { diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 99639476..a5aead0d 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/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 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;