根据devicetoken进行排重

broker/migrate/h2/V17__add_user_session_token_index.sql

@@ -0,0 +1 @@
+alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` );

broker/migrate/mysql/V22__add_user_session_token_index.sql

@@ -0,0 +1 @@
+alter table `t_user_session` ADD INDEX `session_token_index` ( `_token` );

broker/src/main/java/io/moquette/imhandler/UploadDeviceTokenHandler.java

@@ -26,12 +26,12 @@ public class UploadDeviceTokenHandler extends IMHandler<WFCMessage.UploadDeviceT
             session.setAppName(request.getAppName());
             if (request.getPlatform() == ProtoConstants.Platform.Platform_iOS && request.getPushType() == 2) {
                 session.setVoipDeviceToken(request.getDeviceToken());
-//                session.setPushType(0);//如果有voip token，肯定是生产环境的推送
                 m_sessionsStore.updateSessionToken(session, true);
             } else {
                 session.setDeviceToken(request.getDeviceToken());
                 session.setPushType(request.getPushType());
                 m_sessionsStore.updateSessionToken(session, false);
+                m_sessionsStore.cleanDuplatedToken(session.getClientID(), session.getPushType(), session.getDeviceToken(), false);
             }
 
             return ErrorCode.ERROR_CODE_SUCCESS;

broker/src/main/java/io/moquette/persistence/DatabaseStore.java

@@ -1064,10 +1064,45 @@ public class DatabaseStore {
         return null;
     }
 
-    void updateSessionToken(String uid, String cid, String token, boolean voipToken) {
+    void clearOtherSessionToken(String cid, String token, int pushType, boolean voipToken) {
+        if (voipToken) {
+            return;
+        }
+
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+
+            String sql = "update t_user_session set `_token` = ?, `_dt` = ? where `_token` = ? and `_push_type` = ? and `_cid` <> ?";
+
+            statement = connection.prepareStatement(sql);
+
+            int index = 1;
+            statement.setString(index++, "");
+            statement.setLong(index++, System.currentTimeMillis());
+            statement.setString(index++, token);
+            statement.setInt(index++, pushType);
+            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 updateSessionToken(String uid, String cid, String token, int pushType, boolean voipToken) {
         mScheduler.execute(()->{
             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);

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<Map.Entry<String, Session>> 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

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