From 9d23f9c9962a5eb1a598848c790d1afdb96331d7 Mon Sep 17 00:00:00 2001
From: heavyrian2012 <heavyrain.lee@wildfirehat.cn>
Date: Mon, 2 May 2022 11:12:19 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=A2=91=E9=81=93=E8=AE=A2?=
 =?UTF-8?q?=E9=98=85=E5=85=B3=E7=B3=BB=E7=BC=93=E5=AD=98=E9=80=BB=E8=BE=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../persistence/MemoryMessagesStore.java      | 38 ++++++++-----------
 1 file changed, 16 insertions(+), 22 deletions(-)

diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
index ed41c700..6a5c0f88 100755
--- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
@@ -3773,14 +3773,13 @@ public class MemoryMessagesStore implements IMessagesStore {
             return ErrorCode.ERROR_CODE_NOT_RIGHT;
         }
 
-        getChannelListener(channelId);
-
-        if (listen) {
-            listeners.put(channelId, operator);
-        } else {
-            listeners.remove(channelId, operator);
+        try {
+            mWriteLock.lock();
+            databaseStore.updateChannelListener(channelId, operator, listen);
+            listeners.remove(channelId);
+        } finally {
+            mWriteLock.unlock();
         }
-        databaseStore.updateChannelListener(channelId, operator, listen);
 
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
@@ -3790,9 +3789,14 @@ public class MemoryMessagesStore implements IMessagesStore {
         MultiMap<String, String> listeners = hzInstance.getMultiMap(CHANNEL_LISTENERS);
         Collection<String> result = listeners.get(channelId);
         if (result.isEmpty()) {
-            result = databaseStore.getChannelListener(channelId);
-            for (String userId : result) {
-                listeners.put(channelId, userId);
+            try {
+                mWriteLock.lock();
+                result = databaseStore.getChannelListener(channelId);
+                for (String userId : result) {
+                    listeners.put(channelId, userId);
+                }
+            } finally {
+                mWriteLock.unlock();
             }
         }
         return result;
@@ -3821,12 +3825,7 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
 
         Collection<String> channelMembers = getChannelListener(channelId);
-        if (channelMembers == null) {
-            return false;
-        }
-
-        Collection<String> chatroomMembers = getChannelListener(channelId);
-        if (chatroomMembers == null || !chatroomMembers.contains(user)) {
+        if (channelMembers == null || !channelMembers.contains(user)) {
             return false;
         }
 
@@ -3852,12 +3851,7 @@ public class MemoryMessagesStore implements IMessagesStore {
 
     @Override
     public Collection<String> getChannelSubscriber(String channelId) {
-        MultiMap<String, String> channelMembers = m_Server.getHazelcastInstance().getMultiMap(CHANNEL_LISTENERS);
-        if (channelMembers == null) {
-            return new ArrayList<>();
-        }
-
-        return channelMembers.get(channelId);
+        return getChannelListener(channelId);
     }
 
     @Override