diff --git a/broker/migrate/h2/V20__user_session_add_deleted.sql b/broker/migrate/h2/V20__user_session_add_deleted.sql new file mode 100755 index 00000000..c711e696 --- /dev/null +++ b/broker/migrate/h2/V20__user_session_add_deleted.sql @@ -0,0 +1 @@ +alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0; diff --git a/broker/migrate/mysql/V25__user_session_add_deleted.sql b/broker/migrate/mysql/V25__user_session_add_deleted.sql new file mode 100755 index 00000000..c711e696 --- /dev/null +++ b/broker/migrate/mysql/V25__user_session_add_deleted.sql @@ -0,0 +1 @@ +alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0; 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 02e11836..08c35e12 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, false, 0); + ErrorCode errorCode = sessionsStore.loadActiveSession(uid, cid); if (errorCode != ErrorCode.ERROR_CODE_SUCCESS) { sendResponse(response, errorCode, null, base64Response); return true; diff --git a/broker/src/main/java/io/moquette/imhandler/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java index a9c14310..6b066aaa 100755 --- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java @@ -240,7 +240,7 @@ abstract public class IMHandler { WFCMessage.Message.Builder messageBuilder = message.toBuilder(); int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers); - this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType); + mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType)); return notifyReceivers.size(); } @@ -250,7 +250,7 @@ abstract public class IMHandler { message = m_messagesStore.storeMessage(username, clientID, message); WFCMessage.Message.Builder messageBuilder = message.toBuilder(); int pullType = m_messagesStore.getNotifyReceivers(username, messageBuilder, notifyReceivers); - this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType); + mServer.getImBusinessScheduler().execute(() -> this.publisher.publish2Receivers(messageBuilder.build(), notifyReceivers, clientID, pullType)); return notifyReceivers.size(); } diff --git a/broker/src/main/java/io/moquette/imhandler/RouteHandler.java b/broker/src/main/java/io/moquette/imhandler/RouteHandler.java index 030e0305..e6723bea 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, false, request.getPlatform()); + m_sessionsStore.loadActiveSession(fromUser, clientID); } 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 ebe6215b..984419f6 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -16,6 +16,7 @@ import com.hazelcast.core.HazelcastInstance; import com.hazelcast.core.MultiMap; import com.hazelcast.util.StringUtil; import com.xiaoleilu.loServer.model.FriendData; +import io.moquette.server.Server; import io.moquette.spi.ClientSession; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,6 +31,7 @@ import java.sql.*; import java.util.*; import java.util.function.Function; + import static cn.wildfirechat.proto.ProtoConstants.PersistFlag.Transparent; import static io.moquette.persistence.MemoryMessagesStore.USER_STATUS; import static io.moquette.server.Constants.MAX_MESSAGE_QUEUE; @@ -1024,8 +1026,52 @@ public class DatabaseStore { }); } + List getUserActivedSessions(String uid) { + String sql = "select `_cid`, `_package_name`,`_token`,`_voip_token`,`_secret`,`_db_secret`,`_platform`,`_push_type`,`_device_name`,`_device_version`,`_phone_name`,`_language`,`_carrier_name`, `_dt` from t_user_session where `_uid` = ? and `_deleted` = 0"; + Connection connection = null; + PreparedStatement statement = null; + List result = new ArrayList<>(); + try { + connection = DBUtil.getConnection(); + statement = connection.prepareStatement(sql); + statement.setString(1, uid); + + ResultSet resultSet = statement.executeQuery(); + while (resultSet.next()) { + int index = 1; + String cid = resultSet.getString(index++); + + ClientSession clientSession = new ClientSession(cid, Server.getServer().getStore().sessionsStore()); + MemorySessionStore.Session session = new MemorySessionStore.Session(uid, cid, clientSession); + + session.setAppName(resultSet.getString(index++)); + session.setDeviceToken(resultSet.getString(index++)); + session.setVoipDeviceToken(resultSet.getString(index++)); + session.setSecret(resultSet.getString(index++)); + session.setDbSecret(resultSet.getString(index++)); + session.setPlatform(resultSet.getInt(index++)); + session.setPushType(resultSet.getInt(index++)); + + session.setDeviceName(resultSet.getString(index++)); + session.setDeviceVersion(resultSet.getString(index++)); + session.setPhoneName(resultSet.getString(index++)); + session.setLanguage(resultSet.getString(index++)); + session.setCarrierName(resultSet.getString(index++)); + session.setUpdateDt(resultSet.getLong(index++)); + result.add(session); + } + resultSet.close(); + } catch (SQLException e) { + e.printStackTrace(); + Utility.printExecption(LOG, e); + } finally { + DBUtil.closeDB(connection, statement); + } + return result; + } + MemorySessionStore.Session getSession(String uid, String clientId, ClientSession clientSession) { - String sql = "select `_package_name`,`_token`,`_voip_token`,`_secret`,`_db_secret`,`_platform`,`_push_type`,`_device_name`,`_device_version`,`_phone_name`,`_language`,`_carrier_name`, `_dt` from t_user_session where `_uid` = ? and `_cid` = ? limit 1"; + String sql = "select `_package_name`,`_token`,`_voip_token`,`_secret`,`_db_secret`,`_platform`,`_push_type`,`_device_name`,`_device_version`,`_phone_name`,`_language`,`_carrier_name`, `_dt`, `_deleted` from t_user_session where `_uid` = ? and `_cid` = ? limit 1"; Connection connection = null; PreparedStatement statement = null; try { @@ -1051,7 +1097,8 @@ public class DatabaseStore { session.setPhoneName(resultSet.getString(index++)); session.setLanguage(resultSet.getString(index++)); session.setCarrierName(resultSet.getString(index++)); - session.setUpdateDt(resultSet.getLong(index)); + session.setUpdateDt(resultSet.getLong(index++)); + session.setDeleted(resultSet.getInt(index)); return session; } resultSet.close(); @@ -1136,6 +1183,35 @@ public class DatabaseStore { }); } + void updateSessionDeleted(String uid, String cid, int deleted) { + mScheduler.execute(()->{ + Connection connection = null; + PreparedStatement statement = null; + + try { + connection = DBUtil.getConnection(); + + String sql = "update t_user_session set `_deleted` = ? where `_uid` = ? and `_cid` = ?"; + + statement = connection.prepareStatement(sql); + + int index = 1; + statement.setInt(index++, deleted); + 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); + } + }); + } + void updateSessionPlatform(String uid, String cid, int platform) { mScheduler.execute(()->{ Connection connection = null; @@ -1266,13 +1342,19 @@ public class DatabaseStore { 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` <> ?"; + sql = "update t_user_session set `_deleted` = ?, `_token` = ?, `_voip_token` = ?, `_dt` = ? where `_uid`=? and (`_platform` = ? or `_platform` = ?) and `_cid` <> ? and `_deleted` = 0"; } else { - sql = "delete from t_user_session where `_uid`=? and `_platform` = ? and `_cid` <> ?"; + sql = "update t_user_session set `_deleted` = ?, `_token` = ?, `_voip_token` = ?, `_dt` = ? where `_uid`=? and `_platform` = ? and `_cid` <> ? and `_deleted` = 0"; } statement = connection.prepareStatement(sql); int index = 1; + + statement.setInt(index++, 1); + statement.setString(index++, ""); + statement.setString(index++, ""); + statement.setLong(index++, System.currentTimeMillis()); + statement.setString(index++, uid); if (platform == ProtoConstants.Platform.Platform_Windows || platform == ProtoConstants.Platform.Platform_OSX) { @@ -2224,7 +2306,7 @@ public class DatabaseStore { ResultSet rs = null; try { connection = DBUtil.getConnection(); - String sql = "select distinct(`_uid`) from t_user_session"; + String sql = "select distinct(`_uid`) from t_user_session where `_deleted` = 0"; statement = connection.prepareStatement(sql); diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 74c0b37a..0f07a4f9 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -55,6 +55,16 @@ public class MemorySessionStore implements ISessionsStore { private volatile int unReceivedMsgs; + private int deleted; + + public int getDeleted() { + return deleted; + } + + public void setDeleted(int deleted) { + this.deleted = deleted; + } + private MqttVersion mqttVersion = MqttVersion.MQTT_3_1_1; public long getLastActiveTime() { @@ -312,6 +322,11 @@ public class MemorySessionStore implements ISessionsStore { session = databaseStore.createSession(username, clientID, clientSession, platform); } + if (session.getDeleted() > 0) { + session.setDeleted(0); + databaseStore.updateSessionDeleted(username, clientID, 0); + } + if (session.getPlatform() != platform) { session.setPlatform(platform); databaseStore.updateSessionPlatform(username, clientID, platform); @@ -358,34 +373,29 @@ public class MemorySessionStore implements ISessionsStore { @Override - public ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createWhenNoExist, int platform) { + public ErrorCode loadActiveSession(String username, String clientID) { LOG.debug("createNewSession for client <{}>", clientID); Session session = sessions.get(clientID); - if (session != null) { + + if (session != null && session.getDeleted() == 0) { LOG.error("already exists a session for client <{}>, bad condition", clientID); throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); } + if (session != null && session.getDeleted() > 0) { + return ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH; + } ClientSession clientSession = new ClientSession(clientID, this); session = databaseStore.getSession(username, clientID, clientSession); - if (session == null) { - if (!createWhenNoExist) { - return ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH; - } - - session = databaseStore.createSession(username, clientID, clientSession, platform); + if (session == null || session.getDeleted() > 0) { + return ErrorCode.ERROR_CODE_SECRECT_KEY_MISMATCH; } sessions.put(clientID, session); - ConcurrentSkipListSet sessionSet = userSessions.get(username); - if (sessionSet == null) { - sessionSet = new ConcurrentSkipListSet<>(); - userSessions.put(username, sessionSet); - } - sessionSet = userSessions.get(username); + ConcurrentSkipListSet sessionSet = getUserSessionSet(username); sessionSet.add(clientID); return ErrorCode.ERROR_CODE_SUCCESS; @@ -409,12 +419,7 @@ public class MemorySessionStore implements ISessionsStore { session.setUsername(username); sessions.put(clientID, session); - ConcurrentSkipListSet sessionSet = userSessions.get(username); - if (sessionSet == null) { - sessionSet = new ConcurrentSkipListSet<>(); - userSessions.put(username, sessionSet); - } - sessionSet = userSessions.get(username); + ConcurrentSkipListSet sessionSet = getUserSessionSet(username); sessionSet.add(clientID); if (endpoint != null) { @@ -460,14 +465,26 @@ public class MemorySessionStore implements ISessionsStore { } } + private ConcurrentSkipListSet getUserSessionSet(String username) { + ConcurrentSkipListSet sessionSet = userSessions.get(username); + if (sessionSet == null) { + sessionSet = new ConcurrentSkipListSet(); + List ss = databaseStore.getUserActivedSessions(username); + for (Session s : ss) { + sessionSet.add(s.getClientID()); + sessions.put(s.getClientID(), s); + } + userSessions.put(username, sessionSet); + } + + sessionSet = userSessions.get(username); + return sessionSet; + } + @Override public Collection sessionForUser(String username) { - ConcurrentSkipListSet sessionSet = userSessions.get(username); - if (sessionSet == null) { - sessionSet = new ConcurrentSkipListSet(); - userSessions.put(username, sessionSet); - } - sessionSet = userSessions.get(username); + ConcurrentSkipListSet sessionSet = getUserSessionSet(username); + ArrayList out = new ArrayList<>(); for (String clientId : sessionSet ) { @@ -611,12 +628,7 @@ public class MemorySessionStore implements ISessionsStore { LOG.error("Can't find the session for client <{}>", clientID); return; } - ConcurrentSkipListSet sessionSet = userSessions.get(session.username); - if (sessionSet == null) { - sessionSet = new ConcurrentSkipListSet<>(); - userSessions.put(session.username, sessionSet); - } - sessionSet = userSessions.get(session.username); + ConcurrentSkipListSet sessionSet = getUserSessionSet(session.username); sessionSet.remove(clientID); // remove also the messages stored of type QoS1/2 diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 96a807ed..0a2ee81b 100755 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -40,7 +40,7 @@ public interface ISessionsStore { Session createUserSession(String username, String clientID, int platform); - ErrorCode createNewSession(String username, String clientID, boolean cleanSession, boolean createNoExist, int platform); + ErrorCode loadActiveSession(String username, String clientID); 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 2681d049..e0f9b6b2 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -16,6 +16,7 @@ package io.moquette.spi.impl; +import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.proto.WFCMessage; import io.moquette.persistence.RPCCenter; import io.moquette.interception.InterceptHandler; @@ -181,7 +182,11 @@ public class ProtocolProcessor { MqttVersion mqttVersion = MqttVersion.fromProtocolLevel(msg.variableHeader().version()); if (!login(channel, msg, clientId, mqttVersion)) { + MqttConnAckMessage badId = connAck(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD); + + channel.writeAndFlush(badId); channel.close(); + LOG.error("The MQTT login failure. Username={}", payload.userName()); return; } if (!mServer.m_initialized) { @@ -270,9 +275,16 @@ public class ProtocolProcessor { MemorySessionStore.Session session = m_sessionsStore.getSession(clientId); if (session == null) { - m_sessionsStore.createNewSession(msg.payload().userName(), clientId, true, false, 0); + ErrorCode errorCode = m_sessionsStore.loadActiveSession(msg.payload().userName(), clientId); + if (errorCode != ErrorCode.ERROR_CODE_SUCCESS) { + return false; + } session = m_sessionsStore.getSession(clientId); } + + if (session.getDeleted() != 0) { + return false; + } if (session != null && session.getUsername().equals(msg.payload().userName())) { pwd = AES.AESDecrypt(pwd, session.getSecret(), true);