update version to 0.33

broker/migrate/h2/V20__user_session_add_deleted.sql

@@ -0,0 +1 @@
+alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;

broker/migrate/mysql/V25__user_session_add_deleted.sql

@@ -0,0 +1 @@
+alter table `t_user_session` add column `_deleted` tinyint DEFAULT 0;

broker/pom.xml

@@ -5,7 +5,7 @@
         <relativePath>../</relativePath>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.32</version>
+        <version>0.33</version>
     </parent>
 
     <artifactId>moquette-broker</artifactId>

broker/src/main/java/cn/wildfirechat/push/PushServer.java

@@ -57,7 +57,7 @@ public class PushServer {
     }
 
     public void pushMessage(PushMessage pushMessage, String deviceId, String pushContent) {
-        LOG.info("try to delivery push diviceId = {}, pushContent", deviceId, pushContent);
+        LOG.info("try to delivery push diviceId = {}, pushContent = {}", deviceId, pushContent);
         executorService.execute(() ->{
                 try {
                     pushMessageInternel(pushMessage, deviceId, pushContent);

broker/src/main/java/com/xiaoleilu/loServer/action/RouteAction.java

@@ -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

@@ -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

@@ -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

@@ -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;
@@ -638,11 +640,12 @@ public class DatabaseStore {
         String sql = "select  `_from`, `_type`, `_target`, `_line`, `_data`, `_dt` from " + MessageShardingUtil.getMessageTable(messageId) +" where _mid = ? limit 1";
         Connection connection = null;
         PreparedStatement statement = null;
+        ResultSet resultSet = null;
         try {
             connection = DBUtil.getConnection();
             statement = connection.prepareStatement(sql);
             statement.setLong(1, messageId);
-            ResultSet resultSet = statement.executeQuery();
+            resultSet = statement.executeQuery();
             if (resultSet.next()) {
                 WFCMessage.Message.Builder builder = WFCMessage.Message.newBuilder();
                 builder.setMessageId(messageId);
@@ -661,12 +664,11 @@ public class DatabaseStore {
                 WFCMessage.Message message = builder.build();
                 return new MessageBundle(messageId, message.getFromUser(), null, message);
             }
-            resultSet.close();
         } catch (SQLException | IOException e) {
             e.printStackTrace();
             Utility.printExecption(LOG, e);
         } finally {
-            DBUtil.closeDB(connection, statement);
+            DBUtil.closeDB(connection, statement, resultSet);
         }
         return null;
     }
@@ -1024,16 +1026,61 @@ public class DatabaseStore {
         });
     }
 
-    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";
+    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<>();
+        ResultSet resultSet = null;
+        try {
+            connection = DBUtil.getConnection();
+            statement = connection.prepareStatement(sql);
+            statement.setString(1, uid);
+
+            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);
+            }
+        } catch (SQLException e) {
+            e.printStackTrace();
+            Utility.printExecption(LOG, e);
+        } finally {
+            DBUtil.closeDB(connection, statement, resultSet);
+        }
+        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`, `_deleted` from t_user_session where `_uid` = ? and `_cid` = ? limit 1";
+        Connection connection = null;
+        PreparedStatement statement = null;
+        ResultSet resultSet = null;
         try {
             connection = DBUtil.getConnection();
             statement = connection.prepareStatement(sql);
             statement.setString(1, uid);
             statement.setString(2, clientId);
-            ResultSet resultSet = statement.executeQuery();
+            resultSet = statement.executeQuery();
             if (resultSet.next()) {
                 MemorySessionStore.Session session = new MemorySessionStore.Session(uid, clientId, clientSession);
 
@@ -1051,15 +1098,15 @@ 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();
         } catch (SQLException e) {
             e.printStackTrace();
             Utility.printExecption(LOG, e);
         } finally {
-            DBUtil.closeDB(connection, statement);
+            DBUtil.closeDB(connection, statement, resultSet);
         }
         return null;
     }
@@ -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

@@ -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() {
@@ -135,6 +145,9 @@ public class MemorySessionStore implements ISessionsStore {
 
         public void setUpdateDt(long updateDt) {
             this.updateDt = updateDt;
+            if (this.lastActiveTime == 0) {
+                this.lastActiveTime = updateDt;
+            }
         }
 
         private int pushType;
@@ -282,7 +295,7 @@ public class MemorySessionStore implements ISessionsStore {
         while (it.hasNext()) {
             Session session = it.next().getValue();
             if (!session.getClientID().equals(cid) && (session.pushType == pushType && token.equals(session.deviceToken))) {
-                session.deviceToken = token;
+                session.deviceToken = null;
             }
         }
     }
@@ -312,6 +325,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 +376,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 +422,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 +468,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 +631,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

@@ -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

@@ -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);

broker/src/main/java/win/liyufan/im/HttpUtils.java

@@ -52,7 +52,7 @@ public class HttpUtils {
 
             @Override
             public void onResponse(Call call, Response response) throws IOException {
-                LOG.info("POST to {} success with response", url, response.body().string());
+                LOG.info("POST to {} success with response: {}", url, response.body());
             }
         });
     }

common/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.32</version>
+        <version>0.33</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

distribution/pom.xml

@@ -6,7 +6,7 @@
         <!--<relativePath>../pom.xml</relativePath>-->
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.32</version>
+        <version>0.33</version>
     </parent>
 
     <artifactId>distribution</artifactId>

pom.xml

@@ -12,7 +12,7 @@
     <artifactId>wildfirechat-parent</artifactId>
 
     <packaging>pom</packaging>
-    <version>0.32</version>
+    <version>0.33</version>
     <name>Wildfire Chat</name>
     <description>Wildfire Chat Server</description>
     <inceptionYear>2019</inceptionYear>

sdk/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>wildfirechat-parent</artifactId>
         <groupId>cn.wildfirechat</groupId>
-        <version>0.32</version>
+        <version>0.33</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>