anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

解决频道拉取远程消息错误问题

Browse Source
This commit is contained in:
heavyrain2012
2021-12-01 20:22:04 +08:00
parent 74ac13726e
commit e4752c832e
2 changed files with 24 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/persistence/DatabaseStore.java
+12 -5
View File
@@ -784,13 +784,13 @@ public class DatabaseStore {
} }
} }
List<WFCMessage.Message> loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count, Collection<Integer> contentTypes) { List<WFCMessage.Message> loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count, Collection<Integer> contentTypes, String channelOwner) {
List<WFCMessage.Message> messages = new ArrayList<>(); List<WFCMessage.Message> messages = new ArrayList<>();
long[] before = new long[1]; long[] before = new long[1];
before[0] = beforeUid; before[0] = beforeUid;
boolean hasMore = loadRemoteMessagesFromTable(user, conversation, before, count, MessageShardingUtil.getMessageTable(beforeUid), messages, contentTypes); boolean hasMore = loadRemoteMessagesFromTable(user, conversation, before, count, MessageShardingUtil.getMessageTable(beforeUid), messages, contentTypes, channelOwner);
while (messages.size() < count && hasMore) { while (messages.size() < count && hasMore) {
hasMore = loadRemoteMessagesFromTable(user, conversation, before, count - messages.size(), MessageShardingUtil.getMessageTable(beforeUid), messages, contentTypes); hasMore = loadRemoteMessagesFromTable(user, conversation, before, count - messages.size(), MessageShardingUtil.getMessageTable(beforeUid), messages, contentTypes, channelOwner);
} }
int month = 0; int month = 0;
@@ -800,7 +800,7 @@ public class DatabaseStore {
int size = messages.size(); int size = messages.size();
hasMore = true; hasMore = true;
while (size == messages.size() && hasMore) { while (size == messages.size() && hasMore) {
hasMore = loadRemoteMessagesFromTable(user, conversation, before, count - messages.size(), nexTable, messages, contentTypes); hasMore = loadRemoteMessagesFromTable(user, conversation, before, count - messages.size(), nexTable, messages, contentTypes, channelOwner);
} }
if (size < messages.size()) { if (size < messages.size()) {
@@ -811,11 +811,13 @@ public class DatabaseStore {
return messages; return messages;
} }
boolean loadRemoteMessagesFromTable(String user, WFCMessage.Conversation conversation, long[] before, int count, String table, List<WFCMessage.Message> messages, Collection<Integer> contentTypes) { boolean loadRemoteMessagesFromTable(String user, WFCMessage.Conversation conversation, long[] before, int count, String table, List<WFCMessage.Message> messages, Collection<Integer> contentTypes, String channelOwner) {
long beforeUid = before[0]; long beforeUid = before[0];
String sql = "select `_mid`, `_from`, `_type`, `_target`, `_line`, `_data`, `_dt`, `_to` from " + table +" where"; String sql = "select `_mid`, `_from`, `_type`, `_target`, `_line`, `_data`, `_dt`, `_to` from " + table +" where";
if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Private) { if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Private) {
sql += " _type = ? and _line = ? and _mid < ? and ((_target = ? and _from = ?) or (_target = ? and _from = ?))"; sql += " _type = ? and _line = ? and _mid < ? and ((_target = ? and _from = ?) or (_target = ? and _from = ?))";
} else if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Channel && !user.equals(channelOwner)) {
sql += " _type = ? and _line = ? and _mid < ? and _target = ? and ((_from = ? and (_to = '' or _to = ?)) or (_from = ?))";
} else { } else {
sql += " _type = ? and _line = ? and _mid < ? and _target = ?"; sql += " _type = ? and _line = ? and _mid < ? and _target = ?";
} }
@@ -851,7 +853,12 @@ public class DatabaseStore {
statement.setString(index++, user); statement.setString(index++, user);
statement.setString(index++, user); statement.setString(index++, user);
statement.setString(index++, conversation.getTarget()); statement.setString(index++, conversation.getTarget());
} else if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Channel && !user.equals(channelOwner)) {
statement.setString(index++, channelOwner);
statement.setString(index++, user);
statement.setString(index++, user);
} }
statement.setInt(index++, count); statement.setInt(index++, count);
resultSet = statement.executeQuery(); resultSet = statement.executeQuery();
while (resultSet.next()) { while (resultSet.next()) {
broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java
+12 -1
View File
@@ -740,8 +740,19 @@ public class MemoryMessagesStore implements IMessagesStore {
loadMessage = IS_CHATROOM_MESSAGE_REMOTE_HISTORY_MESSAGE; loadMessage = IS_CHATROOM_MESSAGE_REMOTE_HISTORY_MESSAGE;
} }
String channelOwner = null;
if(conversation.getType() == ProtoConstants.ConversationType.ConversationType_Channel) {
IMap<String, WFCMessage.ChannelInfo> mIMap = m_Server.getHazelcastInstance().getMap(CHANNELS);
WFCMessage.ChannelInfo info = mIMap.get(conversation.getTarget());
if(info == null) {
loadMessage = false;
} else {
channelOwner = info.getOwner();
}
}
if (loadMessage) { if (loadMessage) {
messages = databaseStore.loadRemoteMessages(user, conversation, beforeUid, count, contentTypes); messages = databaseStore.loadRemoteMessages(user, conversation, beforeUid, count, contentTypes, channelOwner);
} else { } else {
messages = new ArrayList<>(); messages = new ArrayList<>();
} }