diff --git a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java index 666776ba..a5529eab 100644 --- a/broker/src/main/java/io/moquette/persistence/DatabaseStore.java +++ b/broker/src/main/java/io/moquette/persistence/DatabaseStore.java @@ -784,13 +784,13 @@ public class DatabaseStore { } } - List loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count, Collection contentTypes) { + List loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count, Collection contentTypes, String channelOwner) { List messages = new ArrayList<>(); long[] before = new long[1]; 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) { - 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; @@ -800,7 +800,7 @@ public class DatabaseStore { int size = messages.size(); hasMore = true; 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()) { @@ -811,11 +811,13 @@ public class DatabaseStore { return messages; } - boolean loadRemoteMessagesFromTable(String user, WFCMessage.Conversation conversation, long[] before, int count, String table, List messages, Collection contentTypes) { + boolean loadRemoteMessagesFromTable(String user, WFCMessage.Conversation conversation, long[] before, int count, String table, List messages, Collection contentTypes, String channelOwner) { long beforeUid = before[0]; String sql = "select `_mid`, `_from`, `_type`, `_target`, `_line`, `_data`, `_dt`, `_to` from " + table +" where"; if (conversation.getType() == ProtoConstants.ConversationType.ConversationType_Private) { 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 { 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++, 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); resultSet = statement.executeQuery(); while (resultSet.next()) { diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 20ccdd7a..c45498b1 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -740,8 +740,19 @@ public class MemoryMessagesStore implements IMessagesStore { loadMessage = IS_CHATROOM_MESSAGE_REMOTE_HISTORY_MESSAGE; } + String channelOwner = null; + if(conversation.getType() == ProtoConstants.ConversationType.ConversationType_Channel) { + IMap mIMap = m_Server.getHazelcastInstance().getMap(CHANNELS); + WFCMessage.ChannelInfo info = mIMap.get(conversation.getTarget()); + if(info == null) { + loadMessage = false; + } else { + channelOwner = info.getOwner(); + } + } + if (loadMessage) { - messages = databaseStore.loadRemoteMessages(user, conversation, beforeUid, count, contentTypes); + messages = databaseStore.loadRemoteMessages(user, conversation, beforeUid, count, contentTypes, channelOwner); } else { messages = new ArrayList<>(); }