Skip to content

Commit

Permalink
优化部分逻辑
Browse files Browse the repository at this point in the history
  • Loading branch information
@heavyrain2012
heavyrain2012 committed Apr 2, 2019
1 parent f978136 commit 0f15a38
Show file tree
Hide file tree
Showing 4 changed files with 91 additions and 71 deletions.
97 changes: 74 additions & 23 deletions broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,9 @@ public class MemoryMessagesStore implements IMessagesStore {
private SensitiveFilter mSensitiveFilter;
private volatile long lastUpdateSensitiveTime = 0;

private ConcurrentHashMap<String, Boolean> userGlobalSlientMap = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, Boolean> userPushHiddenDetail = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, Boolean> userConvSlientMap = new ConcurrentHashMap<>();

MemoryMessagesStore(Server server, DatabaseStore databaseStore) {
m_Server = server;
Expand Down Expand Up @@ -1684,6 +1687,28 @@ private List<WFCMessage.UserSettingEntry> loadPersistedUserSettings(String userI
return datas;
}


@Override
public WFCMessage.UserSettingEntry getUserSetting(String userId, int scope, String key) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
MultiMap<String, WFCMessage.UserSettingEntry> userSettingMap = hzInstance.getMultiMap(USER_SETTING);

Collection<WFCMessage.UserSettingEntry> 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<WFCMessage.UserSettingEntry> getUserSetting(String userId, int scope) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
Expand Down Expand Up @@ -1720,29 +1745,6 @@ public long updateUserSettings(String userId, WFCMessage.ModifyUserSettingReq re
}
}

if (request.getScope() == UserSettingScope.kUserSettingConversationSilent || request.getScope() == UserSettingScope.kUserSettingGlobalSilent || request.getScope() == UserSettingScope.kUserSettingHiddenNotificationDetail) {
Collection<MemorySessionStore.Session> 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);
Expand All @@ -1757,9 +1759,58 @@ public long updateUserSettings(String userId, WFCMessage.ModifyUserSettingReq re
}

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();
Expand Down
39 changes: 0 additions & 39 deletions broker/src/main/java/io/moquette/persistence/MemorySessionStore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,10 +46,6 @@ public static class Session implements Comparable<Session>{
private String secret;
private String dbSecret;

private Map<String, String> slientConvs = new ConcurrentHashMap<>();
private boolean globalSlient = false;
private boolean hiddenNotifyDetail = false;

private long lastActiveTime;

private long lastChatroomActiveTime;
Expand Down Expand Up @@ -159,26 +155,6 @@ public Session(String username, String clientID, ClientSession clientSession) {
this.username = username;
}

public boolean isHiddenNotifyDetail() {
return hiddenNotifyDetail;
}

public void setHiddenNotifyDetail(boolean hiddenNotifyDetail) {
this.hiddenNotifyDetail = hiddenNotifyDetail;
}

public Map<String, String> getSlientConvs() {
return slientConvs;
}

public boolean isGlobalSlient() {
return globalSlient;
}

public void setGlobalSlient(boolean globalSlient) {
this.globalSlient = globalSlient;
}

public String getAppName() {
return appName;
}
Expand Down Expand Up @@ -326,21 +302,6 @@ public ErrorCode createNewSession(String username, String clientID, boolean clea
session = databaseStore.createSession(username, clientID, clientSession);
}

List<WFCMessage.UserSettingEntry> 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<WFCMessage.UserSettingEntry> 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<String> sessionSet = userSessions.get(username);
if (sessionSet == null) {
Expand Down
4 changes: 4 additions & 0 deletions broker/src/main/java/io/moquette/spi/IMessagesStore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -161,9 +161,13 @@ public String toString() {
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<WFCMessage.UserSettingEntry> 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);
Expand Down
22 changes: 13 additions & 9 deletions broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -194,16 +194,19 @@ private void publish2Receivers(String sender, int conversationType, String targe
targetClients = m_messagesStore.getChatroomMemberClient(user);
}
for (Session targetSession : sessions) {
//超过7天不活跃的用户忽略
if(System.currentTimeMillis() - targetSession.getLastActiveTime() > 7 * 24 * 60 * 60 * 1000) {
continue;
}

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;
}
Expand All @@ -220,24 +223,24 @@ private void publish2Receivers(String sender, int conversationType, String targe
}

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

Expand Down Expand Up @@ -301,10 +304,11 @@ private void publish2Receivers(String sender, int conversationType, String targe
}

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);
Expand Down

0 comments on commit 0f15a38

Please sign in to comment.