diff --git a/broker/config/wildfirechat.conf b/broker/config/wildfirechat.conf index e176c66c..25fe7168 100755 --- a/broker/config/wildfirechat.conf +++ b/broker/config/wildfirechat.conf @@ -102,6 +102,9 @@ qiniu.bucket_favorite_domain http://cdn2.wildfirechat.cn # 本地媒体服务器配置。 local.media.storage.root media +# 是否支持多端登陆 +server.multi_endpoint false + #********************************************************************* # Push server configuration #********************************************************************* diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java index 6169fe34..02e11836 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java @@ -82,7 +82,7 @@ public class RouteAction extends Action { MemorySessionStore.Session session = sessionsStore.sessionForClientAndUser(uid, cid); if (session == null) { - ErrorCode errorCode = sessionsStore.createNewSession(uid, cid, true, true); + ErrorCode errorCode = sessionsStore.createNewSession(uid, cid, true, false, 0); if (errorCode != ErrorCode.ERROR_CODE_SUCCESS) { sendResponse(response, errorCode, null, base64Response); return true; diff --git a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java index 45617196..ac59a039 100755 --- a/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java +++ b/broker/src/main/java/com/xiaoleilu/loServer/action/admin/GetIMTokenAction.java @@ -41,7 +41,7 @@ public class GetIMTokenAction extends AdminAction { String userId = input.getUserId(); - WFCMessage.GetTokenRequest getTokenRequest = WFCMessage.GetTokenRequest.newBuilder().setUserId(userId).setClientId(input.getClientId()).build(); + WFCMessage.GetTokenRequest getTokenRequest = WFCMessage.GetTokenRequest.newBuilder().setUserId(userId).setClientId(input.getClientId()).setPlatform(input.getPlatform() == null ? 0 : input.getPlatform()).build(); RPCCenter.getInstance().sendRequest(userId, input.getClientId(), IMTopic.GetTokenTopic, getTokenRequest.toByteArray(), userId, TargetEntry.Type.TARGET_TYPE_USER, new RPCCenter.Callback() { @Override public void onSuccess(byte[] result) { diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 922d2436..ac93823b 100755 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -106,6 +106,8 @@ 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 MONGODB_Client_URI = "mongodb.client_uri"; public static final String MONGODB_Database = "mongodb.database"; public static final String MONGODB_Data_Expire_Days = "mongodb.data_expire_days"; diff --git a/broker/src/main/java/io/moquette/imhandler/GetTokenHandler.java b/broker/src/main/java/io/moquette/imhandler/GetTokenHandler.java index c70ad714..557e5813 100644 --- a/broker/src/main/java/io/moquette/imhandler/GetTokenHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/GetTokenHandler.java @@ -20,7 +20,7 @@ import win.liyufan.im.IMTopic; public class GetTokenHandler extends IMHandler { @Override public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.GetTokenRequest request, Qos1PublishHandler.IMCallback callback) { - MemorySessionStore.Session session = m_sessionsStore.createUserSession(fromUser, clientID); + MemorySessionStore.Session session = m_sessionsStore.createUserSession(fromUser, clientID, request.getPlatform()); TokenAuthenticator authenticator = new TokenAuthenticator(); String strToken = authenticator.generateToken(fromUser); String result = strToken + "|" + session.getSecret() + "|" + session.getDbSecret(); diff --git a/broker/src/main/java/io/moquette/imhandler/RouteHandler.java b/broker/src/main/java/io/moquette/imhandler/RouteHandler.java index 47db79c8..030e0305 100644 --- a/broker/src/main/java/io/moquette/imhandler/RouteHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/RouteHandler.java @@ -32,7 +32,7 @@ public class RouteHandler extends IMHandler { ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); boolean isSessionAlreadyStored = clientSession != null; if (!isSessionAlreadyStored) { - m_sessionsStore.createNewSession(fromUser, clientID, true, true); + m_sessionsStore.createNewSession(fromUser, clientID, true, false, request.getPlatform()); } else { m_sessionsStore.updateExistSession(fromUser, clientID, request, true); } diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 14fdaa68..ebe6215b 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -1136,6 +1136,35 @@ public class DatabaseStore { }); } + void updateSessionPlatform(String uid, String cid, int platform) { + mScheduler.execute(()->{ + Connection connection = null; + PreparedStatement statement = null; + + try { + connection = DBUtil.getConnection(); + + String sql = "update t_user_session set `_platform` = ? where `_uid` = ? and `_cid` = ?"; + + statement = connection.prepareStatement(sql); + + int index = 1; + statement.setInt(index++, platform); + statement.setString(index++, uid); + statement.setString(index++, cid); + + int c = statement.executeUpdate(); + LOG.info("Update rows {}", c); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e); + } finally { + DBUtil.closeDB(connection, statement); + } + }); + } + private boolean strEqual(String left, String right) { if (left == right) return true; @@ -1228,23 +1257,65 @@ public class DatabaseStore { }); } - MemorySessionStore.Session createSession(String uid, String clientId, ClientSession clientSession) { + void clearMultiEndpoint(String uid, String clientId, int platform) { + LOG.info("clearMultiEndpoint {}, {}", uid, clientId); + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + String sql; + if (platform == ProtoConstants.Platform.Platform_Windows || platform == ProtoConstants.Platform.Platform_OSX + || platform == ProtoConstants.Platform.Platform_iOS || platform == ProtoConstants.Platform.Platform_Android) { + sql = "delete from t_user_session where `_uid`=? and (`_platform` = ? or `_platform` = ?) and `_cid` <> ?"; + } else { + sql = "delete from t_user_session where `_uid`=? and `_platform` = ? and `_cid` <> ?"; + } + + statement = connection.prepareStatement(sql); + int index = 1; + statement.setString(index++, uid); + + if (platform == ProtoConstants.Platform.Platform_Windows || platform == ProtoConstants.Platform.Platform_OSX) { + statement.setInt(index++, ProtoConstants.Platform.Platform_Windows); + statement.setInt(index++, ProtoConstants.Platform.Platform_OSX); + } else if(platform == ProtoConstants.Platform.Platform_iOS || platform == ProtoConstants.Platform.Platform_Android) { + statement.setInt(index++, ProtoConstants.Platform.Platform_iOS); + statement.setInt(index++, ProtoConstants.Platform.Platform_Android); + } else { + statement.setInt(index++, platform); + } + + statement.setString(index++, clientId); + + int count = statement.executeUpdate(); + LOG.info("Update rows {}", count); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e); + } finally { + DBUtil.closeDB(connection, statement); + } + } + + MemorySessionStore.Session createSession(String uid, String clientId, ClientSession clientSession, int platform) { Connection connection = null; PreparedStatement statement = null; LOG.info("Database create session {},{}", uid, clientId); try { connection = DBUtil.getConnection(); - String sql = "insert into t_user_session (`_uid`,`_cid`,`_secret`,`_db_secret`, `_dt`) values (?,?,?,?,?)"; + String sql = "insert into t_user_session (`_uid`,`_cid`,`_platform`,`_secret`,`_db_secret`, `_dt`) values (?,?,?,?,?,?)"; statement = connection.prepareStatement(sql); int index = 1; MemorySessionStore.Session session = new MemorySessionStore.Session(uid, clientId, clientSession); - + session.setPlatform(platform); statement.setString(index++, uid); statement.setString(index++, clientId); + statement.setInt(index++, platform); session.setSecret(UUID.randomUUID().toString()); diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 20579102..10b4d79f 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -17,8 +17,10 @@ package io.moquette.persistence; import cn.wildfirechat.common.ErrorCode; +import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import com.hazelcast.util.StringUtil; +import io.moquette.BrokerConstants; import io.moquette.server.Constants; import io.moquette.server.Server; import io.moquette.spi.ClientSession; @@ -36,6 +38,8 @@ public class MemorySessionStore implements ISessionsStore { private static int dumy = 1; private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); + private boolean supportMultiEndpoint = false; + public static class Session implements Comparable{ final String clientID; String username; @@ -249,8 +253,14 @@ public class MemorySessionStore implements ISessionsStore { private final Server mServer; private final DatabaseStore databaseStore; public MemorySessionStore(Server server, DatabaseStore databaseStore) { - mServer = server; - this.databaseStore = databaseStore; + mServer = server; + this.databaseStore = databaseStore; + + try { + supportMultiEndpoint = Boolean.parseBoolean(server.getConfig().getProperty(BrokerConstants.SERVER_MULTI_ENDPOINT)); + } catch (Exception e) { + e.printStackTrace(); + } } @Override @@ -292,14 +302,55 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public Session createUserSession(String username, String clientID) { + public Session createUserSession(String username, String clientID, int platform) { LOG.debug("createUserSession for client <{}>, user <{}>", clientID, username); ClientSession clientSession = new ClientSession(clientID, this); Session session = databaseStore.getSession(username, clientID, clientSession); if (session == null) { - session = databaseStore.createSession(username, clientID, clientSession); + session = databaseStore.createSession(username, clientID, clientSession, platform); + } + + if (session.getPlatform() != platform) { + session.setPlatform(platform); + databaseStore.updateSessionPlatform(username, clientID, platform); + } + + + if (!supportMultiEndpoint && platform > 0) { + databaseStore.clearMultiEndpoint(username, clientID, platform); + if (userSessions.get(username) != null) { + Iterator it = userSessions.get(username).iterator(); + while (it.hasNext()) { + String c = it.next(); + if (!clientID.equals(c)) { + Session s = sessions.get(c); + + boolean remove = false; + if (platform == ProtoConstants.Platform.Platform_Android || platform == ProtoConstants.Platform.Platform_iOS) { + if (s.getPlatform() == ProtoConstants.Platform.Platform_Android || s.getPlatform() == ProtoConstants.Platform.Platform_iOS) { + remove = true; + } + } else if(platform == ProtoConstants.Platform.Platform_OSX || platform == ProtoConstants.Platform.Platform_Windows) { + if (s.getPlatform() == ProtoConstants.Platform.Platform_OSX || s.getPlatform() == ProtoConstants.Platform.Platform_Windows) { + remove = true; + } + } else { + if (s.getPlatform() ==platform) { + remove = true; + } + } + + if (remove) { + sessions.remove(c); + mServer.getProcessor().kickoffSession(s); + it.remove(); + } + } + } + } + } return session; @@ -307,7 +358,7 @@ public class MemorySessionStore implements ISessionsStore { @Override - public ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createWhenNoExist) { + public ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createWhenNoExist, int platform) { LOG.debug("createNewSession for client <{}>", clientID); Session session = sessions.get(clientID); @@ -322,10 +373,10 @@ public class MemorySessionStore implements ISessionsStore { if (session == null) { if (!createWhenNoExist) { - return ErrorCode.ERROR_CODE_NOT_EXIST; + return ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH; } - session = databaseStore.createSession(username, clientID, clientSession); + session = databaseStore.createSession(username, clientID, clientSession, platform); } sessions.put(clientID, session); @@ -482,7 +533,9 @@ public class MemorySessionStore implements ISessionsStore { @Override public void dropQueue(String clientID) { - sessions.get(clientID).queue.clear(); + if (sessions.get(clientID) != null) { + sessions.get(clientID).queue.clear(); + } } @Override diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index b962def8..96a807ed 100755 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -38,9 +38,9 @@ public interface ISessionsStore { */ boolean contains(String clientID); - Session createUserSession(String username, String clientID); + Session createUserSession(String username, String clientID, int platform); - ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createNoExist); + ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createNoExist, int platform); ClientSession updateExistSession(String username, String clientID, WFCMessage.RouteRequest endpoint, boolean cleanSession); 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 18b488de..2681d049 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -55,7 +55,24 @@ import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; * * Used by the front facing class ProtocolProcessorBootstrapper. */ + public class ProtocolProcessor { + + + public void kickoffSession(final MemorySessionStore.Session session) { + mServer.getImBusinessScheduler().execute(()->{ + ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID()); + try { + if (descriptor != null) { + processDisconnect(descriptor.getChannel(), true); + } + } catch (InterruptedException e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } + }); + } + private void handleTargetRemovedFromCurrentNode(TargetEntry target) { System.out.println("kickof user " + target); if (target.type == TargetEntry.Type.TARGET_TYPE_USER) { @@ -253,7 +270,7 @@ public class ProtocolProcessor { MemorySessionStore.Session session = m_sessionsStore.getSession(clientId); if (session == null) { - m_sessionsStore.createNewSession(msg.payload().userName(), clientId, true, false); + m_sessionsStore.createNewSession(msg.payload().userName(), clientId, true, false, 0); session = m_sessionsStore.getSession(clientId); } @@ -501,15 +518,12 @@ public class ProtocolProcessor { return; } - boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor); - if (!stillPresent) { - // another descriptor was inserted - LOG.warn("Another descriptor has been inserted. CId={}", clientID); - return; - } + this.connectionDescriptors.removeConnection(existingDescriptor); LOG.info("The DISCONNECT message has been processed. CId={}", clientID); + channel.closeFuture(); + //disconnect the session m_sessionsStore.sessionForClient(clientID).disconnect(clearSession); } diff --git a/broker/src/main/resources/c3p0-config.xml b/broker/src/main/resources/c3p0-config.xml index 512093cf..7cda063a 100755 --- a/broker/src/main/resources/c3p0-config.xml +++ b/broker/src/main/resources/c3p0-config.xml @@ -22,7 +22,7 @@ root - lt123456 + 123456 10 diff --git a/common/src/main/java/cn.wildfirechat.pojos/InputGetToken.java b/common/src/main/java/cn.wildfirechat.pojos/InputGetToken.java index 87e5e1a4..1330bbf2 100755 --- a/common/src/main/java/cn.wildfirechat.pojos/InputGetToken.java +++ b/common/src/main/java/cn.wildfirechat.pojos/InputGetToken.java @@ -3,10 +3,20 @@ package cn.wildfirechat.pojos; public class InputGetToken { private String userId; private String clientId; + private Integer platform; - public InputGetToken(String userId, String clientId) { + public InputGetToken(String userId, String clientId, int platform) { this.userId = userId; this.clientId = clientId; + this.platform = platform; + } + + public Integer getPlatform() { + return platform; + } + + public void setPlatform(Integer platform) { + this.platform = platform; } public String getUserId() { diff --git a/distribution/src/main/resources/wildfirechat.conf b/distribution/src/main/resources/wildfirechat.conf index 318df20f..8a75bfd7 100755 --- a/distribution/src/main/resources/wildfirechat.conf +++ b/distribution/src/main/resources/wildfirechat.conf @@ -95,6 +95,10 @@ qiniu.bucket_favorite_domain http://cdn2.wildfirechat.cn # 本地媒体服务器配置。 local.media.storage.root ./media + +# 是否支持多端登陆 +server.multi_endpoint false + #********************************************************************* # Push server configuration #********************************************************************* diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java index 41d693f7..69dfa1e0 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/Main.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/Main.java @@ -98,7 +98,7 @@ public class Main { System.exit(-1); } - IMResult resultGetToken = UserAdmin.getUserToken(userInfo.getUserId(), "client111"); + IMResult resultGetToken = UserAdmin.getUserToken(userInfo.getUserId(), "client111", ProtoConstants.Platform.Platform_Android); if (resultGetToken != null && resultGetToken.getErrorCode() == ErrorCode.ERROR_CODE_SUCCESS) { System.out.println("get token success: " + resultGetToken.getResult().getToken()); } else { diff --git a/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java b/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java index fd3342e7..7755aa2b 100644 --- a/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java +++ b/sdk/src/main/java/cn/wildfirechat/sdk/UserAdmin.java @@ -2,6 +2,7 @@ package cn.wildfirechat.sdk; import cn.wildfirechat.common.APIPath; import cn.wildfirechat.pojos.*; +import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.sdk.model.IMResult; import cn.wildfirechat.sdk.utilities.AdminHttpUtils; @@ -34,9 +35,9 @@ public class UserAdmin { return AdminHttpUtils.httpJsonPost(path, robot, OutputCreateRobot.class); } - public static IMResult getUserToken(String userId, String clientId) throws Exception { + public static IMResult getUserToken(String userId, String clientId, int platform) throws Exception { String path = APIPath.User_Get_Token; - InputGetToken getToken = new InputGetToken(userId, clientId); + InputGetToken getToken = new InputGetToken(userId, clientId, platform); return AdminHttpUtils.httpJsonPost(path, getToken, OutputGetIMTokenData.class); }