diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index a4f12341..e8bcc761 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -1136,7 +1136,6 @@ public class DatabaseStore { "`_type` = ?," + "`_extra` = ?," + "`_dt` = ?," + - "`_member_count` = ?," + "`_mute` = ?" + ", `_join_type` = ?" + ", `_private_chat` = ?" + @@ -1167,7 +1166,6 @@ public class DatabaseStore { 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()); @@ -1605,12 +1603,10 @@ public class DatabaseStore { ResultSet rs = null; try { connection = DBUtil.getConnection(); - String sql = "select `_gid` from t_group_member where `_mid` = ? and `_type` <> ?"; + String sql = "select `_gid` from t_group_member where `_mid` = ? and `_type` <> 4"; statement = connection.prepareStatement(sql); statement.setString(1, userId); - statement.setInt(2, ProtoConstants.GroupMemberType.GroupMemberType_Removed); - rs = statement.executeQuery(); Set out = new HashSet<>(); @@ -1628,25 +1624,17 @@ public class DatabaseStore { } return null; } - void updateGroupMemberCountDt(final String groupId, final int count, final long dt) { + + void updateGroupMemberDt(final String groupId, final long dt) { Connection connection = null; PreparedStatement statement = null; try { connection = DBUtil.getConnection(); - 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` = ?"; - } + String 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); @@ -1662,6 +1650,29 @@ public class DatabaseStore { } } + void updateGroupMemberCountDt(final String groupId) { + Connection connection = null; + PreparedStatement statement = null; + try { + connection = DBUtil.getConnection(); + + String sql = "update t_group set `_member_count` = (select count(*) from t_group_member where `_gid` = ? and `_type` <> 4 limit 1), `_dt` = ?, `_member_dt` = `_member_dt` + 1 where `_gid` = ?"; + statement = connection.prepareStatement(sql); + statement.setString(1, groupId); + statement.setLong(2, System.currentTimeMillis()); + statement.setString(3, groupId); + + int c = statement.executeUpdate(); + LOG.info("Update rows {}", c); + } catch (SQLException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + Utility.printExecption(LOG, e, RDBS_Exception); + } finally { + DBUtil.closeDB(connection, statement); + } + } + void persistGroupMember(final String groupId, final List memberList) { Connection connection = null; PreparedStatement statement = null; @@ -1724,11 +1735,8 @@ public class DatabaseStore { int removeGroupMember(String groupId, List groupMembers) { Connection connection = null; PreparedStatement statement = null; - PreparedStatement statement2 = null; try { connection = DBUtil.getConnection(); - connection.setAutoCommit(false); - StringBuilder sqlBuilder = new StringBuilder("update t_group_member set `_type` = ?, `_dt` = ? where `_mid` in ("); for (int i = 0; i < groupMembers.size(); i++) { sqlBuilder.append("?"); @@ -1737,9 +1745,7 @@ public class DatabaseStore { } } sqlBuilder.append(")"); - sqlBuilder.append(" and _gid = ?"); - statement = connection.prepareStatement(sqlBuilder.toString()); int index = 1; @@ -1751,36 +1757,14 @@ public class DatabaseStore { statement.setString(index++, memberId); } statement.setString(index++, groupId); - int count = statement.executeUpdate(); LOG.info("Update rows {}", count); - - String sql = "update t_group set `_member_count` = `_member_count` - ?, `_member_dt` = ? , `_dt` = ? where `_gid` = ?"; - - statement2 = connection.prepareStatement(sql); - index = 1; - statement2.setInt(index++, count); - statement2.setLong(index++, current); - statement2.setLong(index++, current); - statement2.setString(index++, groupId); - statement2.executeUpdate(); - connection.commit(); - connection.setAutoCommit(true); } catch (SQLException e) { // TODO Auto-generated catch block - if (connection != null) { - try { - connection.commit(); - connection.setAutoCommit(true); - } catch (SQLException e1) { - e1.printStackTrace(); - } - } e.printStackTrace(); Utility.printExecption(LOG, e, RDBS_Exception); } finally { DBUtil.closeDB(connection, statement); - DBUtil.closeDB(connection, statement2); } return 0; } diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index cb8f3422..ff1614a4 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -1146,26 +1146,11 @@ public class MemoryMessagesStore implements IMessagesStore { } } - for (WFCMessage.GroupMember member : members) { - if (newInviteUsers.contains(member.getMemberId())) { - groupMembers.remove(groupId, member); - } - } - - for (WFCMessage.GroupMember member : memberList) { - groupMembers.put(groupId, member); - } - - int count = 0; - for (WFCMessage.GroupMember member : groupMembers.get(groupId)) { - if (member.getType() != GroupMemberType_Removed) { - count++; - } - } - - mIMap.put(groupId, groupInfo.toBuilder().setMemberUpdateDt(updateDt).setUpdateDt(updateDt).setMemberCount(count).build()); databaseStore.persistGroupMember(groupId, memberList); - databaseStore.updateGroupMemberCountDt(groupId, count, updateDt); + databaseStore.updateGroupMemberCountDt(groupId); + + groupMembers.remove(groupId); + mIMap.evict(groupId); List memberIds = new ArrayList<>(); for (WFCMessage.GroupMember member : memberList) { @@ -1191,12 +1176,12 @@ public class MemoryMessagesStore implements IMessagesStore { void removeGroupMember(String groupId, List memberIds) { HazelcastInstance hzInstance = m_Server.getHazelcastInstance(); databaseStore.removeGroupMember(groupId, memberIds); + databaseStore.updateGroupMemberCountDt(groupId); MultiMap groupMembers = hzInstance.getMultiMap(GROUP_MEMBERS); groupMembers.remove(groupId); IMap mIMap = hzInstance.getMap(GROUPS_MAP); mIMap.evict(groupId); - } void removeFavGroup(String groupId, List memberIds) { @@ -1384,8 +1369,6 @@ public class MemoryMessagesStore implements IMessagesStore { } } - - WFCMessage.GroupInfo.Builder newInfoBuilder = oldInfo.toBuilder(); if (modifyType == Modify_Group_Name) @@ -1407,7 +1390,8 @@ public class MemoryMessagesStore implements IMessagesStore { newInfoBuilder.setUpdateDt(System.currentTimeMillis()); - mIMap.put(groupId, newInfoBuilder.build()); + databaseStore.persistGroupInfo(newInfoBuilder.build()); + mIMap.evict(groupId); if (modifyType == Modify_Group_Mute) { if (newInfoBuilder.getMute() > 0) { @@ -1488,7 +1472,7 @@ public class MemoryMessagesStore implements IMessagesStore { groupMembers.remove(groupId, member); member = member.toBuilder().setAlias(alias).setUpdateDt(updateDt).build(); databaseStore.persistGroupMember(groupId, Arrays.asList(member)); - databaseStore.updateGroupMemberCountDt(groupId, -1, updateDt); + databaseStore.updateGroupMemberDt(groupId, updateDt); groupMembers.put(groupId, member); mIMap.set(groupId, groupInfo.toBuilder().setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build()); @@ -1692,11 +1676,10 @@ public class MemoryMessagesStore implements IMessagesStore { member = member.toBuilder().setType(type == 0 ? ProtoConstants.GroupMemberType.GroupMemberType_Normal : ProtoConstants.GroupMemberType.GroupMemberType_Manager).setUpdateDt(updateDt).build(); databaseStore.persistGroupMember(groupId, Arrays.asList(member)); groupMembers.put(groupId, member); -// userList.remove(member.getMemberId()); } } - - mIMap.put(groupId, groupInfo.toBuilder().setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build()); + databaseStore.persistGroupInfo(groupInfo.toBuilder().setUpdateDt(updateDt).setMemberUpdateDt(updateDt).build()); + mIMap.evict(groupId); callbackGroupMemberEvent(operator, groupId, userList, ProtoConstants.GroupMemberUpdateEventType.Group_Member_Event_Type_Update, (type == 0 ? ProtoConstants.GroupMemberType.GroupMemberType_Normal : ProtoConstants.GroupMemberType.GroupMemberType_Manager) + ""); return ErrorCode.ERROR_CODE_SUCCESS;