From 772940e50d3a9c0983bd327655604205a5a9fee6 Mon Sep 17 00:00:00 2001
From: imhao183 <wildfirechat86@gmail.com>
Date: Thu, 31 Dec 2020 20:14:23 +0800
Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E6=B8=85=E9=99=A4=E6=95=B0?=
 =?UTF-8?q?=E6=8D=AE=E5=BA=93=E5=90=8E=E6=8B=89=E5=8F=96=E5=8E=86=E5=8F=B2?=
 =?UTF-8?q?=E6=B6=88=E6=81=AF=E6=97=B6=EF=BC=8C=E6=97=A0=E6=B3=95=E8=8E=B7?=
 =?UTF-8?q?=E5=8F=96=E8=87=AA=E5=B7=B1=E5=8F=91=E9=80=81=E6=B6=88=E6=81=AF?=
 =?UTF-8?q?=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../moquette/persistence/MemoryMessagesStore.java   | 13 ++++++++++---
 .../io/moquette/persistence/MemorySessionStore.java | 10 ++++++++++
 2 files changed, 20 insertions(+), 3 deletions(-)

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<Long, Long> 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;
         }