diff --git a/broker/migrate/h2/V17__add_user_session_token_index.sql b/broker/migrate/h2/V17__add_user_session_token_index.sql new file mode 100755 index 00000000..395dbab1 --- /dev/null +++ b/broker/migrate/h2/V17__add_user_session_token_index.sql @@ -0,0 +1 @@ +alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` ); diff --git a/broker/migrate/mysql/V22__add_user_session_token_index.sql b/broker/migrate/mysql/V22__add_user_session_token_index.sql new file mode 100755 index 00000000..395dbab1 --- /dev/null +++ b/broker/migrate/mysql/V22__add_user_session_token_index.sql @@ -0,0 +1 @@ +alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` ); diff --git a/broker/src/main/java/io/moquette/imhandler/UploadDeviceTokenHandler.java b/broker/src/main/java/io/moquette/imhandler/UploadDeviceTokenHandler.java index 56ad2a57..2d89c2b3 100644 --- a/broker/src/main/java/io/moquette/imhandler/UploadDeviceTokenHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/UploadDeviceTokenHandler.java @@ -26,12 +26,12 @@ public class UploadDeviceTokenHandler extends IMHandler{ Connection connection = null; PreparedStatement statement = null; + + clearOtherSessionToken(cid, token, pushType, voipToken); try { connection = DBUtil.getConnection(); @@ -1076,12 +1111,15 @@ public class DatabaseStore { if (voipToken) { sql = "update t_user_session set `_voip_token` = ?, `_dt` = ? where `_uid` = ? and `_cid` = ?"; } else { - sql = "update t_user_session set `_token` = ?, `_dt` = ? where `_uid` = ? and `_cid` = ?"; + sql = "update t_user_session set `_token` = ?, `_push_type` = ?, `_dt` = ? where `_uid` = ? and `_cid` = ?"; } statement = connection.prepareStatement(sql); int index = 1; statement.setString(index++, token); + if (!voipToken) { + statement.setInt(index++, pushType); + } statement.setLong(index++, System.currentTimeMillis()); statement.setString(index++, uid); statement.setString(index++, cid); diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index b9d6c256..20579102 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -16,7 +16,9 @@ package io.moquette.persistence; +import cn.wildfirechat.common.ErrorCode; import cn.wildfirechat.proto.WFCMessage; +import com.hazelcast.util.StringUtil; import io.moquette.server.Constants; import io.moquette.server.Server; import io.moquette.spi.ClientSession; @@ -26,7 +28,6 @@ import io.moquette.spi.ISessionsStore; import io.netty.handler.codec.mqtt.MqttVersion; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import cn.wildfirechat.common.ErrorCode; import java.util.*; import java.util.concurrent.*; @@ -261,6 +262,21 @@ public class MemorySessionStore implements ISessionsStore { return session; } + @Override + public void cleanDuplatedToken(String cid, int pushType, String token, boolean isVoip) { + if (StringUtil.isNullOrEmpty(token) || isVoip) { + return; + } + + Iterator> it = sessions.entrySet().iterator(); + while (it.hasNext()) { + Session session = it.next().getValue(); + if (!session.getClientID().equals(cid) && (session.pushType == pushType && token.equals(session.deviceToken))) { + session.deviceToken = token; + } + } + } + @Override public void initStore() { } @@ -272,7 +288,7 @@ public class MemorySessionStore implements ISessionsStore { @Override public void updateSessionToken(Session session, boolean voip) { - databaseStore.updateSessionToken(session.getUsername(), session.getClientID(), voip ? session.getVoipDeviceToken() : session.getDeviceToken(), voip); + databaseStore.updateSessionToken(session.getUsername(), session.getClientID(), voip ? session.getVoipDeviceToken() : session.getDeviceToken(), session.getPushType(), voip); } @Override diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index d6153bc8..b962def8 100755 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -46,6 +46,8 @@ public interface ISessionsStore { Session getSession(String clientID); + void cleanDuplatedToken(String cid, int pushType, String token, boolean isVoip); + void updateSessionToken(Session session, boolean voip); /** * @param clientID