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

优化用户session

Browse Source
This commit is contained in:
heavyrain2012
2019-11-27 10:30:39 +08:00
parent 9aaaa71ce9
commit f9d5c861ce
9 changed files with 151 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/migrate/h2/V20__user_session_add_deleted.sql
Executable
+1
View File
@@ -0,0 +1 @@
alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;
broker/migrate/mysql/V25__user_session_add_deleted.sql
Executable
+1
View File
@@ -0,0 +1 @@
alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;
broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java
+1 -1
View File
@@ -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;
broker/src/main/java/io/moquette/imhandler/IMHandler.java
+2 -2
View File
@@ -240,7 +240,7 @@ abstract public class IMHandler<T> {
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<T> {
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();
}
broker/src/main/java/io/moquette/imhandler/RouteHandler.java
+1 -1
View File
@@ -32,7 +32,7 @@ public class RouteHandler extends IMHandler<WFCMessage.RouteRequest> {
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);
}
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+87 -5
View File
@@ -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<MemorySessionStore.Session> 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<MemorySessionStore.Session> 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);
broker/src/main/java/io/moquette/persistence/MemorySessionStore.java
+44 -32
View File
@@ -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<String> sessionSet = userSessions.get(username);
if (sessionSet == null) {
sessionSet = new ConcurrentSkipListSet<>();
userSessions.put(username, sessionSet);
}
sessionSet = userSessions.get(username);
ConcurrentSkipListSet<String> 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<String> sessionSet = userSessions.get(username);
if (sessionSet == null) {
sessionSet = new ConcurrentSkipListSet<>();
userSessions.put(username, sessionSet);
}
sessionSet = userSessions.get(username);
ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(username);
sessionSet.add(clientID);
if (endpoint != null) {
@@ -460,14 +465,26 @@ public class MemorySessionStore implements ISessionsStore {
}
}
private ConcurrentSkipListSet<String> getUserSessionSet(String username) {
ConcurrentSkipListSet<String> sessionSet = userSessions.get(username);
if (sessionSet == null) {
sessionSet = new ConcurrentSkipListSet<String>();
List<Session> 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<Session> sessionForUser(String username) {
ConcurrentSkipListSet<String> sessionSet = userSessions.get(username);
if (sessionSet == null) {
sessionSet = new ConcurrentSkipListSet<String>();
userSessions.put(username, sessionSet);
}
sessionSet = userSessions.get(username);
ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(username);
ArrayList<Session> 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<String> sessionSet = userSessions.get(session.username);
if (sessionSet == null) {
sessionSet = new ConcurrentSkipListSet<>();
userSessions.put(session.username, sessionSet);
}
sessionSet = userSessions.get(session.username);
ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(session.username);
sessionSet.remove(clientID);
// remove also the messages stored of type QoS1/2
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+1 -1
View File
@@ -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);
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+13 -1
View File
@@ -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);