优化session处理

This reverts commit ab7b20871a.

.gitignore

@@ -56,3 +56,6 @@ broker/media
 common/target
 server/target
 sdk/target
+moments/target
+monitor
+wildfire_jmeter

broker/pom.xml

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

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

@@ -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, request.getPlatform());
+        MemorySessionStore.Session session = m_sessionsStore.updateOrCreateUserSession(fromUser, clientID, request.getPlatform());
         TokenAuthenticator authenticator = new TokenAuthenticator();
         String strToken = authenticator.generateToken(fromUser);
         String result = strToken + "|" + session.getSecret() + "|" + session.getDbSecret();

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

@@ -20,7 +20,7 @@ import java.util.List;
 public class UserSearchHandler extends IMHandler<WFCMessage.SearchUserRequest> {
     @Override
     public ErrorCode action(ByteBuf ackPayload, String clientID, String fromUser, boolean isAdmin, WFCMessage.SearchUserRequest request, Qos1PublishHandler.IMCallback callback) {
-        List<WFCMessage.User> users = m_messagesStore.searchUser(request.getKeyword(), request.getFuzzy() > 0, request.getPage());
+        List<WFCMessage.User> users = m_messagesStore.searchUser(request.getKeyword(), request.getFuzzy(), request.getPage());
         WFCMessage.SearchUserResult.Builder builder = WFCMessage.SearchUserResult.newBuilder();
         builder.addAllEntry(users);
         byte[] data = builder.build().toByteArray();

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

@@ -35,6 +35,7 @@ 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;
+import static cn.wildfirechat.proto.ProtoConstants.SearchUserType.*;
 
 public class DatabaseStore {
     private static final Logger LOG = LoggerFactory.getLogger(DatabaseStore.class);
@@ -148,8 +149,12 @@ public class DatabaseStore {
         }
         return null;
     }
-
-    List<WFCMessage.User> searchUserFromDB(String keyword, boolean buzzy, int page) {
+    //searchType
+    // 0 模糊匹配displayName, 精确匹配账户名和电话号码
+    // 1 精确匹配账户名和电话号码
+    // 2 精确匹配账户名
+    // 3 精确匹配电话号码
+    List<WFCMessage.User> searchUserFromDB(String keyword, int searchType, int page) {
         Connection connection = null;
         PreparedStatement statement = null;
         ResultSet rs = null;
@@ -169,18 +174,29 @@ public class DatabaseStore {
                 ", `_social`" +
                 ", `_extra`" +
                 ", `_dt` from t_user";
-            if (buzzy) {
-                sql += " where (`_display_name` like ? or `_name` = ? or `_mobile` = ?) ";
-            } else {
-                sql += " where (`_name` = ? or `_mobile` = ?) ";
+            switch (searchType) {
+                case SearchUserType_Name_Mobile:
+                    sql += " where (`_name` = ? or `_mobile` = ?) ";
+                    break;
+                case SearchUserType_Name:
+                    sql += " where `_name` = ? ";
+                    break;
+                case SearchUserType_Mobile:
+                    sql += " where `_mobile` = ? ";
+                    break;
+                case SearchUserType_General:
+                default:
+                    sql += " where (`_display_name` like ? or `_name` = ? or `_mobile` = ?) ";
+                    break;
             }
 
+
             sql += " and _type <> 2"; //can search normal user(0) and robot(1), can not search things
 
-            if (buzzy) {
-                sql += " limit 20";
-            } else {
+            if (searchType == SearchUserType_Name_Mobile || searchType == SearchUserType_Name || searchType == SearchUserType_Mobile) {
                 sql += " limit 1";
+            } else {
+                sql += " limit 20";
             }
             
             if (page > 0) {
@@ -190,12 +206,24 @@ public class DatabaseStore {
 
             statement = connection.prepareStatement(sql);
             int index = 1;
-            if (buzzy) {
-                statement.setString(index++, "%" + keyword + "%");
+
+            switch (searchType) {
+                case SearchUserType_Name_Mobile:
+                    statement.setString(index++, keyword);
+                    statement.setString(index++, keyword);
+                    break;
+                case SearchUserType_Name:
+                case SearchUserType_Mobile:
+                    statement.setString(index++, keyword);
+                    break;
+                case SearchUserType_General:
+                default:
+                    statement.setString(index++, "%" + keyword + "%");
+                    statement.setString(index++, keyword);
+                    statement.setString(index++, keyword);
+                    break;
             }
 
-            statement.setString(index++, keyword);
-            statement.setString(index++, keyword);
 
             rs = statement.executeQuery();
             while (rs.next()) {
@@ -1011,78 +1039,76 @@ public class DatabaseStore {
     }
 
     void persistGroupInfo(final WFCMessage.GroupInfo groupInfo) {
-        mScheduler.execute(()->{
-            Connection connection = null;
-            PreparedStatement statement = null;
-            try {
-                connection = DBUtil.getConnection();
-                String sql = "insert into t_group (`_gid`" +
-                    ", `_name`" +
-                    ", `_portrait`" +
-                    ", `_owner`" +
-                    ", `_type`" +
-                    ", `_extra`" +
-                    ", `_dt`" +
-                    ", `_member_count`" +
-                    ", `_mute`" +
-                    ", `_join_type`" +
-                    ", `_private_chat`" +
-                    ", `_searchable`" +
-                    ", `_member_dt`) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" +
-                    " ON DUPLICATE KEY UPDATE " +
-                    "`_name` = ?," +
-                    "`_portrait` = ?," +
-                    "`_owner` = ?," +
-                    "`_type` = ?," +
-                    "`_extra` = ?," +
-                    "`_dt` = ?," +
-                    "`_member_count` = ?," +
-                    "`_mute` = ?" +
-                    ", `_join_type` = ?" +
-                    ", `_private_chat` = ?" +
-                    ", `_searchable` = ?, " +
-                    "`_member_dt` = ?";
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
+            String sql = "insert into t_group (`_gid`" +
+                ", `_name`" +
+                ", `_portrait`" +
+                ", `_owner`" +
+                ", `_type`" +
+                ", `_extra`" +
+                ", `_dt`" +
+                ", `_member_count`" +
+                ", `_mute`" +
+                ", `_join_type`" +
+                ", `_private_chat`" +
+                ", `_searchable`" +
+                ", `_member_dt`) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" +
+                " ON DUPLICATE KEY UPDATE " +
+                "`_name` = ?," +
+                "`_portrait` = ?," +
+                "`_owner` = ?," +
+                "`_type` = ?," +
+                "`_extra` = ?," +
+                "`_dt` = ?," +
+                "`_member_count` = ?," +
+                "`_mute` = ?" +
+                ", `_join_type` = ?" +
+                ", `_private_chat` = ?" +
+                ", `_searchable` = ?, " +
+                "`_member_dt` = ?";
 
 
-                statement = connection.prepareStatement(sql);
+            statement = connection.prepareStatement(sql);
 
-                int index = 1;
-                statement.setString(index++, groupInfo.getTargetId());
-                statement.setString(index++, groupInfo.getName());
-                statement.setString(index++, groupInfo.getPortrait());
-                statement.setString(index++, groupInfo.getOwner());
-                statement.setInt(index++, groupInfo.getType());
-                statement.setString(index++, groupInfo.getExtra());
-                statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
-                statement.setInt(index++, groupInfo.getMemberCount());
-                statement.setInt(index++, groupInfo.getMute());
-                statement.setInt(index++, groupInfo.getJoinType());
-                statement.setInt(index++, groupInfo.getPrivateChat());
-                statement.setInt(index++, groupInfo.getSearchable());
-                statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
+            int index = 1;
+            statement.setString(index++, groupInfo.getTargetId());
+            statement.setString(index++, groupInfo.getName());
+            statement.setString(index++, groupInfo.getPortrait());
+            statement.setString(index++, groupInfo.getOwner());
+            statement.setInt(index++, groupInfo.getType());
+            statement.setString(index++, groupInfo.getExtra());
+            statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
+            statement.setInt(index++, groupInfo.getMemberCount());
+            statement.setInt(index++, groupInfo.getMute());
+            statement.setInt(index++, groupInfo.getJoinType());
+            statement.setInt(index++, groupInfo.getPrivateChat());
+            statement.setInt(index++, groupInfo.getSearchable());
+            statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
 
-                statement.setString(index++, groupInfo.getName());
-                statement.setString(index++, groupInfo.getPortrait());
-                statement.setString(index++, groupInfo.getOwner());
-                statement.setInt(index++, groupInfo.getType());
-                statement.setString(index++, groupInfo.getExtra());
-                statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
-                statement.setInt(index++, groupInfo.getMemberCount());
-                statement.setInt(index++, groupInfo.getMute());
-                statement.setInt(index++, groupInfo.getJoinType());
-                statement.setInt(index++, groupInfo.getPrivateChat());
-                statement.setInt(index++, groupInfo.getSearchable());
-                statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
-                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);
-            }
-        });
+            statement.setString(index++, groupInfo.getName());
+            statement.setString(index++, groupInfo.getPortrait());
+            statement.setString(index++, groupInfo.getOwner());
+            statement.setInt(index++, groupInfo.getType());
+            statement.setString(index++, groupInfo.getExtra());
+            statement.setLong(index++, groupInfo.getUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
+            statement.setInt(index++, groupInfo.getMemberCount());
+            statement.setInt(index++, groupInfo.getMute());
+            statement.setInt(index++, groupInfo.getJoinType());
+            statement.setInt(index++, groupInfo.getPrivateChat());
+            statement.setInt(index++, groupInfo.getSearchable());
+            statement.setLong(index++, groupInfo.getMemberUpdateDt() == 0 ? System.currentTimeMillis() : groupInfo.getUpdateDt());
+            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);
+        }
     }
 
     List<MemorySessionStore.Session> getUserActivedSessions(String uid) {
@@ -1532,39 +1558,37 @@ public class DatabaseStore {
         return null;
     }
     void updateGroupMemberCountDt(final String groupId, final int count, final long dt) {
-        mScheduler.execute(()->{
-            Connection connection = null;
-            PreparedStatement statement = null;
-            try {
-                connection = DBUtil.getConnection();
+        Connection connection = null;
+        PreparedStatement statement = null;
+        try {
+            connection = DBUtil.getConnection();
 
-                String sql ;
+            String sql ;
 
-                if (count >= 0) {
-                    sql = "update t_group set `_member_count` = ?, `_member_dt` = ? , `_dt` = ? where `_gid` = ?";
-                } else {
-                    sql = "update t_group set `_member_dt` = ?, `_dt` = ? where `_gid` = ?";
-                }
-                statement = connection.prepareStatement(sql);
-
-                int index = 1;
-                if (count >=0) {
-                    statement.setInt(index++, count);
-                }
-                statement.setLong(index++, dt);
-                statement.setLong(index++, dt);
-                statement.setString(index++, groupId);
-
-                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);
+            if (count >= 0) {
+                sql = "update t_group set `_member_count` = ?, `_member_dt` = ? , `_dt` = ? where `_gid` = ?";
+            } else {
+                sql = "update t_group set `_member_dt` = ?, `_dt` = ? where `_gid` = ?";
             }
-        });
+            statement = connection.prepareStatement(sql);
+
+            int index = 1;
+            if (count >=0) {
+                statement.setInt(index++, count);
+            }
+            statement.setLong(index++, dt);
+            statement.setLong(index++, dt);
+            statement.setString(index++, groupId);
+
+            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 persistGroupMember(final String groupId, final List<WFCMessage.GroupMember> memberList) {

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

@@ -42,7 +42,6 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.*;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
@@ -112,7 +111,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     private final DatabaseStore databaseStore;
     private ConcurrentHashMap<String, Long> userMaxPullSeq = new ConcurrentHashMap<>();
 
-    private AtomicReference<SensitiveFilter> mSensitiveFilter = new AtomicReference<>();
+    private SensitiveFilter mSensitiveFilter;
     private volatile long lastUpdateSensitiveTime = 0;
 
     private ConcurrentHashMap<String, Boolean> userGlobalSlientMap = new ConcurrentHashMap<>();
@@ -148,7 +147,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                 }
             }
             Set<String> sensitiveWords = databaseStore.getSensitiveWord();
-            mSensitiveFilter.set(new SensitiveFilter(sensitiveWords));
+            mSensitiveFilter = new SensitiveFilter(sensitiveWords);
         }
     }
 
@@ -353,6 +352,9 @@ public class MemoryMessagesStore implements IMessagesStore {
 
                         size += bundle.getMessage().getSerializedSize();
                         if (size >= 512 * 1024) { //3M
+                            if(builder.getMessageCount() == 0){
+                                builder.addMessage(bundle.getMessage());
+                            }
                             break;
                         }
                         builder.addMessage(bundle.getMessage());
@@ -687,7 +689,11 @@ public class MemoryMessagesStore implements IMessagesStore {
         List<WFCMessage.GroupMember> tmp = new ArrayList<>();
         ArrayList<String> newInviteUsers = new ArrayList<>();
         for (WFCMessage.GroupMember member : memberList) {
-            member = member.toBuilder().setType(ProtoConstants.GroupMemberType.GroupMemberType_Normal).setUpdateDt(updateDt).setAlias("").build();
+            if (member.getMemberId().equals(groupInfo.getOwner())) {
+                member = member.toBuilder().setType(GroupMemberType_Owner).setUpdateDt(updateDt).setAlias("").build();
+            } else {
+                member = member.toBuilder().setType(GroupMemberType_Normal).setUpdateDt(updateDt).setAlias("").build();
+            }
             tmp.add(member);
             newInviteUsers.add(member.getMemberId());
         }
@@ -1492,8 +1498,8 @@ public class MemoryMessagesStore implements IMessagesStore {
         return getUserInfo(userId);
     }
     @Override
-    public List<WFCMessage.User> searchUser(String keyword, boolean buzzy, int page) {
-        return databaseStore.searchUserFromDB(keyword, buzzy, page);
+    public List<WFCMessage.User> searchUser(String keyword, int searchType, int page) {
+        return databaseStore.searchUserFromDB(keyword, searchType, page);
     }
 
     @Override
@@ -1992,7 +1998,7 @@ public class MemoryMessagesStore implements IMessagesStore {
         } else {
             state = 0;
         }
-        
+
         HazelcastInstance hzInstance = m_Server.getHazelcastInstance();
         MultiMap<String, FriendData> friendsMap = hzInstance.getMultiMap(USER_FRIENDS);
 
@@ -2495,7 +2501,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     @Override
     public Set<String> handleSensitiveWord(String message) {
         updateSensitiveWord();
-        return mSensitiveFilter.get().getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH);
+        return mSensitiveFilter.getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH);
     }
 
     @Override
@@ -2638,7 +2644,7 @@ public class MemoryMessagesStore implements IMessagesStore {
         }
         return UUIDGenerator.getUUID();
     }
-    
+
     @Override
     public void storeRetained(Topic topic, StoredMessage storedMessage) {
         LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID());

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

@@ -328,15 +328,24 @@ public class MemorySessionStore implements ISessionsStore {
     }
 
     @Override
-    public Session createUserSession(String username, String clientID, int platform) {
+    public Session updateOrCreateUserSession(String username, String clientID, int platform) {
         LOG.debug("createUserSession for client <{}>, user <{}>", clientID, username);
 
+        Session session = sessions.get(clientID);
+
+        if (session != null && !session.username.equals(username)) {
+            if (userSessions.get(username) != null) {
+                userSessions.get(username).remove(clientID);
+            }
+        }
         ClientSession clientSession = new ClientSession(clientID, this);
-        Session session = databaseStore.getSession(username, clientID, clientSession);
+        session = databaseStore.getSession(username, clientID, clientSession);
 
         if (session == null) {
             session = databaseStore.createSession(username, clientID, clientSession, platform);
         }
+        sessions.put(clientID, session);
+
 
         if (session.getDeleted() > 0) {
             session.setDeleted(0);

broker/src/main/java/io/moquette/spi/IMessagesStore.java

@@ -139,7 +139,7 @@ public interface IMessagesStore {
     WFCMessage.User getUserInfo(String userId);
     WFCMessage.User getUserInfoByName(String name);
     WFCMessage.User getUserInfoByMobile(String mobile);
-    List<WFCMessage.User> searchUser(String keyword, boolean buzzy, int page);
+    List<WFCMessage.User> searchUser(String keyword, int searchType, int page);
 
     boolean updateSystemSetting(int id, String value, String desc);
     SystemSettingPojo getSystemSetting(int id);

broker/src/main/java/io/moquette/spi/ISessionsStore.java

@@ -38,7 +38,7 @@ public interface ISessionsStore {
      */
     boolean contains(String clientID);
 
-    Session createUserSession(String username, String clientID, int platform);
+    Session updateOrCreateUserSession(String username, String clientID, int platform);
 
     ErrorCode loadActiveSession(String username, String clientID);
 

broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java

@@ -179,15 +179,16 @@ public class Qos1PublishHandler extends QosPublishHandler {
                             MemorySessionStore.Session session = m_sessionStore.getSession(clientID);
                             if (session != null && session.getUsername().equals(fromUser)) {
                                 if (data.length > 7*1024 && session.getMqttVersion().protocolLevel() >= MqttVersion.Wildfire_1.protocolLevel()) {
-                                    try (
-                                        ByteArrayOutputStream out = new ByteArrayOutputStream();
-                                        GZIPOutputStream gzip = new GZIPOutputStream(out)
-                                    ) {
+                                    ByteArrayOutputStream out = new ByteArrayOutputStream();
+                                    GZIPOutputStream gzip;
+                                    try {
+                                        gzip = new GZIPOutputStream(out);
                                         gzip.write(data);
-                                        data = out.toByteArray();
+                                        gzip.close();
                                     } catch (Exception e) {
                                         e.printStackTrace();
                                     }
+                                    data = out.toByteArray();
                                     code = (byte)ErrorCode.ERROR_CODE_SUCCESS_GZIPED.code;
                                 }
 

common/pom.xml

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

common/src/main/java/cn/wildfirechat/proto/ProtoConstants.java

@@ -132,6 +132,12 @@ public class ProtoConstants {
         int Group_Max_Member_Count = 1;
     }
 
+    public interface SearchUserType {
+        int SearchUserType_General = 0;
+        int SearchUserType_Name_Mobile = 1;
+        int SearchUserType_Name = 2;
+        int SearchUserType_Mobile = 3;
+    }
 
 
 

distribution/pom.xml

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

pom.xml

@@ -12,7 +12,7 @@
     <artifactId>wildfirechat-parent</artifactId>
 
     <packaging>pom</packaging>
-    <version>0.34</version>
+    <version>0.36</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.34</version>
+        <version>0.36</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>