限制多端登陆
This commit is contained in:
@@ -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";
|
||||
|
||||
@@ -20,7 +20,7 @@ import win.liyufan.im.IMTopic;
|
||||
public class GetTokenHandler extends IMHandler<WFCMessage.GetTokenRequest> {
|
||||
@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();
|
||||
|
||||
@@ -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, true);
|
||||
m_sessionsStore.createNewSession(fromUser, clientID, true, false, request.getPlatform());
|
||||
} else {
|
||||
m_sessionsStore.updateExistSession(fromUser, clientID, request, true);
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -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<Session>{
|
||||
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<String> 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
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user