From 12d22e78bfa4d852f24387d3e0a348d0c6cd7656 Mon Sep 17 00:00:00 2001 From: heavyrain2012 Date: Thu, 21 Nov 2019 20:41:55 +0800 Subject: [PATCH] =?UTF-8?q?=E6=A0=B9=E6=8D=AEdevicetoken=E8=BF=9B=E8=A1=8C?= =?UTF-8?q?=E6=8E=92=E9=87=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../h2/V17__add_user_session_token_index.sql | 1 + .../V22__add_user_session_token_index.sql | 1 + .../imhandler/UploadDeviceTokenHandler.java | 2 +- .../moquette/persistence/DatabaseStore.java | 42 ++++++++++++++++++- .../persistence/MemorySessionStore.java | 20 ++++++++- .../java/io/moquette/spi/ISessionsStore.java | 2 + 6 files changed, 63 insertions(+), 5 deletions(-) create mode 100755 broker/migrate/h2/V17__add_user_session_token_index.sql create mode 100755 broker/migrate/mysql/V22__add_user_session_token_index.sql 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