diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index 0317a7c9..87c2e329 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -142,6 +142,9 @@ local.media.storage.root media # 是否支持多端登陆 server.multi_endpoint false +# 多平台连接状态通知（仅当multi_endpoint为false时有效） +server.multi_platform_notification false + #********************************************************************* # Push server configuration #********************************************************************* diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 0f991ccc..b72fed83 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -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"; diff --git a/broker/src/main/java/io/moquette/imhandler/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java index 41c269e1..15d84576 100755 --- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java @@ -65,6 +65,9 @@ abstract public class IMHandler { protected static String actionName; + public static MessagesPublisher getPublisher() { + return publisher; + } public IMHandler() { try { diff --git a/broker/src/main/java/io/moquette/imhandler/PutUserSettingHandler.java b/broker/src/main/java/io/moquette/imhandler/PutUserSettingHandler.java index 26f6ed11..02bf1a3c 100644 --- a/broker/src/main/java/io/moquette/imhandler/PutUserSettingHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/PutUserSettingHandler.java @@ -18,8 +18,7 @@ import win.liyufan.im.IMTopic; public class PutUserSettingHandler extends IMHandler { @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; } } diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 05055f8e..9397754c 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -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; } diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index b7b32bf9..581ef757 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -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; + } } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index a545fbaa..dcc85977 100755 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -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(); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 606e54e8..4e9c98d2 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -131,6 +131,8 @@ public interface IMessagesStore { ErrorCode getUserInfo(List 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 getUserStatusList(); diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index eca5ae83..b3648169 100755 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -149,4 +149,6 @@ public interface ISessionsStore { int getSecondPhaseAckPendingMessages(String clientID); void cleanSession(String clientID); + + boolean isMultiEndpointSupported(); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 35728368..d03b439b 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -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); diff --git a/broker/src/main/java/win/liyufan/im/UserSettingScope.java b/broker/src/main/java/win/liyufan/im/UserSettingScope.java index 68b9454d..ae6a4c8c 100644 --- a/broker/src/main/java/win/liyufan/im/UserSettingScope.java +++ b/broker/src/main/java/win/liyufan/im/UserSettingScope.java @@ -11,6 +11,7 @@ public interface UserSettingScope { int kUserSettingConversationSync = 7; int kUserSettingMyChannels = 8; int kUserSettingListenedChannels = 9; + int kUserSettingPCOnline = 10; int kUserSettingCustomBegin = 1000; } diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf index 0a57dd4b..bc24541c 100755 --- a/distribution/src/main/resources/wildfirechat.conf +++ b/distribution/src/main/resources/wildfirechat.conf @@ -136,6 +136,9 @@ local.media.storage.root ./media # 是否支持多端登陆 server.multi_endpoint false +# 多平台连接状态通知（仅当multi_endpoint为false时有效） +server.multi_platform_notification false + #********************************************************************* # Push server configuration #*********************************************************************