anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加PC在线状态通知开关

Browse Source
This commit is contained in:
imhao183
2020-04-05 21:14:39 +08:00
parent 38f42697cb
commit da011f4957
12 changed files with 143 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/config/wildfirechat.conf
+3
View File
@@ -142,6 +142,9 @@ local.media.storage.root media
# 是否支持多端登陆
server.multi_endpoint false
# 多平台连接状态通知（仅当multi_endpoint为false时有效）
server.multi_platform_notification false
#*********************************************************************
# Push server configuration
#*********************************************************************
broker/src/main/java/io/moquette/BrokerConstants.java
+1
View File
@@ -110,6 +110,7 @@ public final class BrokerConstants {
public static final String MESSAGE_Forward_Url = "message.forward.url";
public static final String SERVER_MULTI_ENDPOINT = "server.multi_endpoint";
public static final String SERVER_MULTI_PLATFROM_NOTIFICATION = "server.multi_platform_notification";
public static final String MONGODB_Client_URI = "mongodb.client_uri";
public static final String MONGODB_Database = "mongodb.database";
broker/src/main/java/io/moquette/imhandler/IMHandler.java
+3
View File
@@ -65,6 +65,9 @@ abstract public class IMHandler<T> {
protected static String actionName;
public static MessagesPublisher getPublisher() {
return publisher;
}
public IMHandler() {
try {
broker/src/main/java/io/moquette/imhandler/PutUserSettingHandler.java
+1 -2
View File
@@ -18,8 +18,7 @@ import win.liyufan.im.IMTopic;
public class PutUserSettingHandler extends IMHandler<WFCMessage.ModifyUserSettingReq> {
@Override
public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.ModifyUserSettingReq request, Qos1PublishHandler.IMCallback callback) {
long timestamp = m_messagesStore.updateUserSettings(fromUser, request);
publisher.publishNotification(IMTopic.NotifyUserSettingTopic, fromUser, timestamp);
m_messagesStore.updateUserSettings(fromUser, request);
return ErrorCode.ERROR_CODE_SUCCESS;
}
}
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+110
View File
@@ -26,6 +26,7 @@ import com.xiaoleilu.loServer.model.FriendData;
import cn.wildfirechat.pojos.InputOutputUserBlockStatus;
import cn.wildfirechat.common.ErrorCode;
import io.moquette.BrokerConstants;
import io.moquette.imhandler.IMHandler;
import io.moquette.server.Constants;
import io.moquette.server.Server;
import io.moquette.spi.IMatchingCondition;
@@ -55,10 +56,12 @@ import static cn.wildfirechat.proto.ProtoConstants.GroupMemberType.*;
import static cn.wildfirechat.proto.ProtoConstants.ModifyChannelInfoType.*;
import static cn.wildfirechat.proto.ProtoConstants.ModifyGroupInfoType.*;
import static cn.wildfirechat.proto.ProtoConstants.PersistFlag.Transparent;
import static cn.wildfirechat.proto.ProtoConstants.Platform.*;
import static io.moquette.BrokerConstants.*;
import static io.moquette.server.Constants.MAX_CHATROOM_MESSAGE_QUEUE;
import static io.moquette.server.Constants.MAX_MESSAGE_QUEUE;
import static win.liyufan.im.MyInfoType.*;
import static win.liyufan.im.UserSettingScope.kUserSettingPCOnline;
public class MemoryMessagesStore implements IMessagesStore {
private static final String MESSAGES_MAP = "messages_map";
@@ -125,6 +128,7 @@ public class MemoryMessagesStore implements IMessagesStore {
private long mFriendRejectDuration = 30 * 24 * 60 * 60 * 1000;
private long mFriendRequestExpiration = 7 * 24 * 60 * 60 * 1000;
private boolean mMultiPlatformNotification = false;
private boolean mDisableStrangerChat = false;
private long mChatroomParticipantIdleTime = 900000;
@@ -145,6 +149,15 @@ public class MemoryMessagesStore implements IMessagesStore {
Utility.printExecption(LOG, e);
printMissConfigLog(MESSAGE_Disable_Stranger_Chat, mDisableStrangerChat + "");
}
try {
mMultiPlatformNotification = Boolean.parseBoolean(m_Server.getConfig().getProperty(BrokerConstants.SERVER_MULTI_PLATFROM_NOTIFICATION, "false"));
} catch (Exception e) {
e.printStackTrace();
Utility.printExecption(LOG, e);
printMissConfigLog(SERVER_MULTI_PLATFROM_NOTIFICATION, mDisableStrangerChat + "");
}
try {
mDisableSearch = Boolean.parseBoolean(m_Server.getConfig().getProperty(BrokerConstants.FRIEND_Disable_Search));
} catch (Exception e) {
@@ -1559,6 +1572,102 @@ public class MemoryMessagesStore implements IMessagesStore {
}
@Override
public void updateUserOnlineSetting(MemorySessionStore.Session session, boolean online) {
if(!mMultiPlatformNotification) {
return;
}
if (m_Server.getStore().sessionsStore().isMultiEndpointSupported()) {
return;
}
if (m_Server.isShutdowning()) {
return;
}
if (session.getPlatform() == Platform_Linux || session.getPlatform() == Platform_Windows || session.getPlatform() == Platform_OSX) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("PC").setValue(online ? (System.currentTimeMillis() + "|" + session.getPlatform() + "|" + session.getClientID() + "|" + session.getPhoneName()) : "").build());
} else if(session.getPlatform() == Platform_WEB) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("Web").setValue(online ? (System.currentTimeMillis() + "|" + session.getPlatform() + "|" + session.getClientID() + "|" + session.getPhoneName()) : "").build());
} else if(session.getPlatform() == Platform_WX) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("WX").setValue(online ? (System.currentTimeMillis() + "|" + session.getPlatform() + "|" + session.getClientID() + "|" + session.getPhoneName()) : "").build());
} else {
boolean isPCOnline = false;
String pcClientid = null;
String pcName = null;
int pcPlatform = 0;
boolean isWebOnline = false;
String webClientid = null;
String webName = null;
boolean isWXOnline = false;
String wxClientid = null;
String wxName = null;
for (MemorySessionStore.Session s : m_Server.getStore().sessionsStore().sessionForUser(session.username)) {
if (s.getDeleted() != 0 || !m_Server.getConnectionsManager().isConnected(s.getClientID())) {
continue;
}
switch (s.getPlatform()) {
case Platform_Linux:
case Platform_Windows:
case Platform_OSX:
isPCOnline = true;
pcClientid = s.getClientID();
pcName = s.getPhoneName();
pcPlatform = s.getPlatform();
break;
case Platform_WEB:
isWebOnline = true;
webClientid = s.getClientID();
webName = s.getPhoneName();
break;
case Platform_WX:
isWXOnline = true;
wxClientid = s.getClientID();
wxName = s.getPhoneName();
break;
default:
break;
}
}
WFCMessage.UserSettingEntry pcentry = getUserSetting(session.getUsername(), kUserSettingPCOnline, "PC");
if (isPCOnline) {
if (pcentry == null || StringUtil.isNullOrEmpty(pcentry.getValue())) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("PC").setValue(online ? (System.currentTimeMillis() + "|" + pcPlatform + "|" + pcClientid + "|" + pcName) : "").build());
}
} else {
if (pcentry != null && !StringUtil.isNullOrEmpty(pcentry.getValue())) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("PC").setValue("").build());
}
}
WFCMessage.UserSettingEntry webentry = getUserSetting(session.getUsername(), kUserSettingPCOnline, "Web");
if (isWebOnline) {
if (webentry == null || StringUtil.isNullOrEmpty(webentry.getValue())) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("Web").setValue(online ? (System.currentTimeMillis() + "|" + Platform_WEB + "|" + webClientid + "|" + webName) : "").build());
}
} else {
if (webentry != null && !StringUtil.isNullOrEmpty(webentry.getValue())) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("Web").setValue("").build());
}
}
WFCMessage.UserSettingEntry wxentry = getUserSetting(session.getUsername(), kUserSettingPCOnline, "WX");
if (isWXOnline) {
if (wxentry == null || StringUtil.isNullOrEmpty(wxentry.getValue())) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("WX").setValue(online ? (System.currentTimeMillis() + "|" + Platform_WX + "|" + wxClientid + "|" + wxName) : "").build());
}
} else {
if (wxentry != null && !StringUtil.isNullOrEmpty(wxentry.getValue())) {
updateUserSettings(session.username, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("WX").setValue(online ? (System.currentTimeMillis() + "|" + Platform_WX + "|" + wxClientid + "|" + wxName) : "").build());
}
}
}
}
@Override
public ErrorCode modifyUserStatus(String userId, int status) {
HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
@@ -2501,6 +2610,7 @@ public class MemoryMessagesStore implements IMessagesStore {
userGlobalSlientMap.remove(userId);
userConvSlientMap.remove(userId);
userPushHiddenDetail.remove(userId);
IMHandler.getPublisher().publishNotification(IMTopic.NotifyUserSettingTopic, userId, updateDt);
return updateDt;
}
broker/src/main/java/io/moquette/persistence/MemorySessionStore.java
+4
View File
@@ -680,4 +680,8 @@ public class MemorySessionStore implements ISessionsStore {
// TODO this missing last step breaks the junit test
sessions.remove(clientID);
}
@Override
public boolean isMultiEndpointSupported() {
return supportMultiEndpoint;
}
}
broker/src/main/java/io/moquette/server/Server.java
+8
View File
@@ -82,6 +82,8 @@ public class Server {
private ServerAcceptor m_acceptor;
private boolean m_shutdowning = false;
public volatile boolean m_initialized;
private ProtocolProcessor m_processor;
@@ -415,9 +417,15 @@ public class Server {
m_processor.onRpcMsg(fromUser, clientId, message, messageId, from, request, isAdmin);
}
public boolean isShutdowning() {
return m_shutdowning;
}
public void stopServer() {
System.out.println("Server will flush data to db before shutting down, please wait 5 seconds!");
LOG.info("Unbinding server from the configured ports");
m_shutdowning = true;
m_acceptor.close();
LOG.trace("Stopping MQTT protocol processor");
m_processorBootstrapper.shutdown();
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+2
View File
@@ -131,6 +131,8 @@ public interface IMessagesStore {
ErrorCode getUserInfo(List<WFCMessage.UserRequest> requestList, WFCMessage.PullUserResult.Builder builder);
ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) throws Exception;
void updateUserOnlineSetting(MemorySessionStore.Session session, boolean online);
ErrorCode modifyUserStatus(String userId, int status);
int getUserStatus(String userId);
List<InputOutputUserBlockStatus> getUserStatusList();
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+2
View File
@@ -149,4 +149,6 @@ public interface ISessionsStore {
int getSecondPhaseAckPendingMessages(String clientID);
void cleanSession(String clientID);
boolean isMultiEndpointSupported();
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+5
View File
@@ -244,6 +244,7 @@ public class ProtocolProcessor {
if(session != null) {
session.refreshLastActiveTime();
forwardOnlineStatusEvent(payload.userName(), clientId, session.getPlatform(), UserOnlineStatus.ONLINE);
m_messagesStore.updateUserOnlineSetting(session, true);
} else {
forwardOnlineStatusEvent(payload.userName(), clientId, ProtoConstants.Platform.Platform_UNSET, UserOnlineStatus.ONLINE);
}
@@ -559,6 +560,9 @@ public class ProtocolProcessor {
String username = NettyUtils.userName(channel);
MemorySessionStore.Session session = m_sessionsStore.getSession(clientID);
if(session != null) {
m_messagesStore.updateUserOnlineSetting(session, false);
}
if(session != null) {
forwardOnlineStatusEvent(username, clientID, session.getPlatform(), UserOnlineStatus.LOGOUT);
}
@@ -605,6 +609,7 @@ public class ProtocolProcessor {
if(session != null) {
session.refreshLastActiveTime();
forwardOnlineStatusEvent(username, clientID, session.getPlatform(), clearSession ? UserOnlineStatus.LOGOUT : UserOnlineStatus.OFFLINE);
m_messagesStore.updateUserOnlineSetting(session, false);
}
ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel);
broker/src/main/java/win/liyufan/im/UserSettingScope.java
+1
View File
@@ -11,6 +11,7 @@ public interface UserSettingScope {
int kUserSettingConversationSync = 7;
int kUserSettingMyChannels = 8;
int kUserSettingListenedChannels = 9;
int kUserSettingPCOnline = 10;
int kUserSettingCustomBegin = 1000;
}
distribution/src/main/resources/wildfirechat.conf
+3
View File
@@ -136,6 +136,9 @@ local.media.storage.root ./media
# 是否支持多端登陆
server.multi_endpoint false
# 多平台连接状态通知（仅当multi_endpoint为false时有效）
server.multi_platform_notification false
#*********************************************************************
# Push server configuration
#*********************************************************************