From 0f15a3840e277b2cb25587fee188edbca8762bef Mon Sep 17 00:00:00 2001 From: heavyrain2012 Date: Tue, 2 Apr 2019 22:55:29 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=83=A8=E5=88=86=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../persistence/MemoryMessagesStore.java | 97 ++++++++++++++----- .../persistence/MemorySessionStore.java | 39 -------- .../java/io/moquette/spi/IMessagesStore.java | 4 + .../moquette/spi/impl/MessagesPublisher.java | 22 +++-- 4 files changed, 91 insertions(+), 71 deletions(-) diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 2363b9a2..1f7a0008 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -112,6 +112,9 @@ public class MemoryMessagesStore implements IMessagesStore { private SensitiveFilter mSensitiveFilter; private volatile long lastUpdateSensitiveTime = 0; + private ConcurrentHashMap userGlobalSlientMap = new ConcurrentHashMap<>(); + private ConcurrentHashMap userPushHiddenDetail = new ConcurrentHashMap<>(); + private ConcurrentHashMap userConvSlientMap = new ConcurrentHashMap<>(); MemoryMessagesStore(Server server, DatabaseStore databaseStore) { m_Server = server; @@ -1684,6 +1687,28 @@ public class MemoryMessagesStore implements IMessagesStore { return datas; } + + @Override + public WFCMessage.UserSettingEntry getUserSetting(String userId, int scope, String key) { + HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); + MultiMap userSettingMap = hzInstance.getMultiMap(USER_SETTING); + + Collection entries = userSettingMap.get(userId); + if (entries == null || entries.size() == 0) { + entries = loadPersistedUserSettings(userId, userSettingMap); + } + + if (entries != null) { + for (WFCMessage.UserSettingEntry entry : entries) { + if (entry.getScope() == scope && (key== null || key.equals(entry.getKey()))) { + return entry; + } + } + } + + return null; + } + @Override public List getUserSetting(String userId, int scope) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); @@ -1720,29 +1745,6 @@ public class MemoryMessagesStore implements IMessagesStore { } } - if (request.getScope() == UserSettingScope.kUserSettingConversationSilent || request.getScope() == UserSettingScope.kUserSettingGlobalSilent || request.getScope() == UserSettingScope.kUserSettingHiddenNotificationDetail) { - Collection sessions = m_Server.getStore().sessionsStore().sessionForUser(userId); - for (MemorySessionStore.Session targetSession : sessions) { - if (StringUtil.isNullOrEmpty(request.getValue()) || request.getValue().equals("0")) { - if(request.getScope() == UserSettingScope.kUserSettingConversationSilent) { - targetSession.getSlientConvs().remove(request.getKey()); - } else if(request.getScope() == UserSettingScope.kUserSettingGlobalSilent) { - targetSession.setGlobalSlient(false); - } else if(request.getScope() == UserSettingScope.kUserSettingHiddenNotificationDetail) { - targetSession.setHiddenNotifyDetail(false); - } - } else { - if(request.getScope() == UserSettingScope.kUserSettingConversationSilent) { - targetSession.getSlientConvs().put(request.getKey(), request.getValue()); - } else if(request.getScope() == UserSettingScope.kUserSettingGlobalSilent) { - targetSession.setGlobalSlient(true); - } else if(request.getScope() == UserSettingScope.kUserSettingHiddenNotificationDetail) { - targetSession.setHiddenNotifyDetail(true); - } - } - } - } - long updateDt = System.currentTimeMillis(); WFCMessage.UserSettingEntry settingEntry = WFCMessage.UserSettingEntry.newBuilder().setScope(request.getScope()).setKey(request.getKey()).setValue(request.getValue()).setUpdateDt(updateDt).build(); databaseStore.persistUserSetting(userId, settingEntry); @@ -1757,9 +1759,58 @@ public class MemoryMessagesStore implements IMessagesStore { } userSettingMap.put(userId, settingEntry); + userGlobalSlientMap.remove(userId); + userConvSlientMap.remove(userId); return updateDt; } + @Override + public boolean getUserGlobalSlient(String userId) { + Boolean slient = userGlobalSlientMap.get(userId); + if (slient == null) { + WFCMessage.UserSettingEntry entry = getUserSetting(userId, UserSettingScope.kUserSettingGlobalSilent, null); + if (entry == null || !entry.getValue().equals("1")) { + slient = false; + } else { + slient = true; + } + userGlobalSlientMap.put(userId, slient); + } + return slient; + } + + @Override + public boolean getUserPushHiddenDetail(String userId) { + Boolean hidden = userPushHiddenDetail.get(userId); + if (hidden == null) { + WFCMessage.UserSettingEntry entry = getUserSetting(userId, UserSettingScope.kUserSettingGlobalSilent, null); + if (entry == null || !entry.getValue().equals("1")) { + hidden = false; + } else { + hidden = true; + } + userPushHiddenDetail.put(userId, hidden); + } + return hidden; + } + + @Override + public boolean getUserConversationSlient(String userId, WFCMessage.Conversation conversation) { + String key = userId + "|" + conversation.getType() + "|" + conversation.getTarget() + "|" + conversation.getLine(); + Boolean slient = userConvSlientMap.get(key); + if (slient == null) { + String convSlientKey = conversation.getType() + "-" + conversation.getLine() + "-" + conversation.getTarget(); + WFCMessage.UserSettingEntry entry = getUserSetting(userId, UserSettingScope.kUserSettingConversationSilent, convSlientKey); + if (entry == null || !entry.getValue().equals("1")) { + slient = false; + } else { + slient = true; + } + userConvSlientMap.put(userId, slient); + } + return slient; + } + @Override public ErrorCode createChannel(String operator, WFCMessage.ChannelInfo channelInfo) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index cac25b93..6b0abc1d 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -46,10 +46,6 @@ public class MemorySessionStore implements ISessionsStore { private String secret; private String dbSecret; - private Map slientConvs = new ConcurrentHashMap<>(); - private boolean globalSlient = false; - private boolean hiddenNotifyDetail = false; - private long lastActiveTime; private long lastChatroomActiveTime; @@ -159,26 +155,6 @@ public class MemorySessionStore implements ISessionsStore { this.username = username; } - public boolean isHiddenNotifyDetail() { - return hiddenNotifyDetail; - } - - public void setHiddenNotifyDetail(boolean hiddenNotifyDetail) { - this.hiddenNotifyDetail = hiddenNotifyDetail; - } - - public Map getSlientConvs() { - return slientConvs; - } - - public boolean isGlobalSlient() { - return globalSlient; - } - - public void setGlobalSlient(boolean globalSlient) { - this.globalSlient = globalSlient; - } - public String getAppName() { return appName; } @@ -326,21 +302,6 @@ public class MemorySessionStore implements ISessionsStore { session = databaseStore.createSession(username, clientID, clientSession); } - List settings = mServer.getStore().messagesStore().getUserSetting(username, UserSettingScope.kUserSettingConversationSilent); - for (WFCMessage.UserSettingEntry entry : settings) { - if (!StringUtil.isNullOrEmpty(entry.getValue()) && !entry.getValue().equals("0")) { - session.slientConvs.put(entry.getKey(), entry.getValue()); - break; - } - } - - List settings2 = mServer.getStore().messagesStore().getUserSetting(username, UserSettingScope.kUserSettingGlobalSilent); - for (WFCMessage.UserSettingEntry entry : settings2) { - if (!StringUtil.isNullOrEmpty(entry.getValue()) && !entry.getValue().equals("0")) { - session.globalSlient = true; - break; - } - } sessions.put(clientID, session); ConcurrentSkipListSet sessionSet = userSessions.get(username); if (sessionSet == null) { diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 29802f49..982c1f8a 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -161,9 +161,13 @@ public interface IMessagesStore { ErrorCode handleQuitChatroom(String userId, String clientId, String chatroomId); ErrorCode getUserSettings(String userId, long version, WFCMessage.GetUserSettingResult.Builder builder); + WFCMessage.UserSettingEntry getUserSetting(String userId, int scope, String key); List getUserSetting(String userId, int scope); long updateUserSettings(String userId, WFCMessage.ModifyUserSettingReq request); + boolean getUserGlobalSlient(String userId); + boolean getUserPushHiddenDetail(String userId); + boolean getUserConversationSlient(String userId, WFCMessage.Conversation conversation); ErrorCode createChannel(String operator, WFCMessage.ChannelInfo channelInfo); ErrorCode modifyChannelInfo(String operator, String channelId, int modifyType, String value); diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 318e76aa..bece73e7 100755 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -194,6 +194,7 @@ public class MessagesPublisher { targetClients = m_messagesStore.getChatroomMemberClient(user); } for (Session targetSession : sessions) { + //超过7天不活跃的用户忽略 if(System.currentTimeMillis() - targetSession.getLastActiveTime() > 7 * 24 * 60 * 60 * 1000) { continue; } @@ -201,9 +202,11 @@ public class MessagesPublisher { if (exceptClientId != null && exceptClientId.equals(targetSession.getClientSession().clientID)) { continue; } + if (targetSession.getClientID() == null) { continue; } + if (pullType == ProtoConstants.PullType.Pull_ChatRoom && !targetClients.contains(targetSession.getClientID())) { continue; } @@ -220,24 +223,24 @@ public class MessagesPublisher { } boolean isSlient; - String slientKey = null; if (pullType == ProtoConstants.PullType.Pull_ChatRoom) { isSlient = true; } else { isSlient = false; + WFCMessage.Conversation conversation; if (conversationType == ProtoConstants.ConversationType.ConversationType_Private) { - slientKey = conversationType + "-" + line + "-" + sender; + conversation = WFCMessage.Conversation.newBuilder().setType(conversationType).setLine(line).setTarget(sender).build(); } else { - slientKey = conversationType + "-" + line + "-" + target; + conversation = WFCMessage.Conversation.newBuilder().setType(conversationType).setLine(line).setTarget(target).build(); } - if (targetSession.getSlientConvs().containsKey(slientKey)) { - LOG.info("The conversation {} is slient", slientKey); + if (m_messagesStore.getUserConversationSlient(user, conversation)) { + LOG.info("The conversation {}-{}-{} is slient", conversation.getType(), conversation.getTarget(), conversation.getLine()); isSlient = true; } - if (targetSession.isGlobalSlient()) { - LOG.info("The user is global sliented"); + if (m_messagesStore.getUserGlobalSlient(user)) { + LOG.info("The user {} is global sliented", user); isSlient = true; } @@ -301,10 +304,11 @@ public class MessagesPublisher { } if (isSlient) { - LOG.info("The conversation {} is slient", slientKey); + LOG.info("The conversation is slient"); continue; } - boolean isHiddenDetail = targetSession.isHiddenNotifyDetail(); + + boolean isHiddenDetail = m_messagesStore.getUserPushHiddenDetail(user); if(!nameLoaded) { senderName = getUserDisplayName(sender);