diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index ae6796f7..7d853a24 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -560,9 +560,13 @@ public class MemoryMessagesStore implements IMessagesStore { } boolean noRoaming = false; - if (pullType == ProtoConstants.PullType.Pull_Normal && fromMessageId == 0 && !IS_MESSAGE_ROAMING) { - noRoaming = true; + if (pullType == ProtoConstants.PullType.Pull_Normal && fromMessageId == 0) { + if(!IS_MESSAGE_ROAMING) { + noRoaming = true; + } + session.setPullHistoryMsg(true); } + while (true) { Map.Entry entry = maps.higherEntry(current); if (entry == null) { @@ -574,7 +578,7 @@ public class MemoryMessagesStore implements IMessagesStore { MessageBundle bundle = mIMap.get(targetMessageId); if (bundle != null) { - if (exceptClientId == null || !exceptClientId.equals(bundle.getFromClientId()) || !user.equals(bundle.getFromUser())) { + if (exceptClientId == null || session.isPullHistoryMsg() || !exceptClientId.equals(bundle.getFromClientId()) || !user.equals(bundle.getFromUser())) { if (pullType == ProtoConstants.PullType.Pull_ChatRoom) { if (!bundle.getMessage().getConversation().getTarget().equals(chatroomId)) { @@ -608,6 +612,9 @@ public class MemoryMessagesStore implements IMessagesStore { if (lastEntry != null) { head = lastEntry.getKey(); } + if (pullType == ProtoConstants.PullType.Pull_Normal && current == head) { + session.setPullHistoryMsg(false); + } } finally { mReadLock.unlock(); } diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 71bd00b9..d407760e 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -57,6 +57,8 @@ public class MemorySessionStore implements ISessionsStore { private int deleted; + private boolean pullHistoryMsg; + public int getDeleted() { return deleted; } @@ -139,6 +141,14 @@ public class MemorySessionStore implements ISessionsStore { this.carrierName = carrierName; } + public boolean isPullHistoryMsg() { + return pullHistoryMsg; + } + + public void setPullHistoryMsg(boolean pullHistoryMsg) { + this.pullHistoryMsg = pullHistoryMsg; + } + public long getUpdateDt() { return updateDt; }