From e366962c65f693ea329b36d46f267ee003af1aa8 Mon Sep 17 00:00:00 2001 From: "heavyrain.lee" Date: Sun, 1 Mar 2020 14:45:44 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E5=AE=9A=E5=90=91=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E6=8B=89=E5=8F=96=E8=BF=9C=E7=A8=8B=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E6=97=B6=E7=9A=84=E9=94=99=E8=AF=AF=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../h2/V23__alter_message_add_to_column.sql | 2 + .../V28__alter_message_add_to_column.sql | 38 +++++++++++++ .../imhandler/LoadRemoteMessagesHandler.java | 6 +++ .../moquette/persistence/DatabaseStore.java | 54 ++++++++++++++----- .../persistence/MemoryMessagesStore.java | 10 ++++ .../java/io/moquette/spi/IMessagesStore.java | 1 + 6 files changed, 98 insertions(+), 13 deletions(-) create mode 100755 broker/migrate/h2/V23__alter_message_add_to_column.sql create mode 100755 broker/migrate/mysql/V28__alter_message_add_to_column.sql diff --git a/broker/migrate/h2/V23__alter_message_add_to_column.sql b/broker/migrate/h2/V23__alter_message_add_to_column.sql new file mode 100755 index 00000000..17860857 --- /dev/null +++ b/broker/migrate/h2/V23__alter_message_add_to_column.sql @@ -0,0 +1,2 @@ +alter table t_messages add column `_to` varchar(64) DEFAULT NULL; + diff --git a/broker/migrate/mysql/V28__alter_message_add_to_column.sql b/broker/migrate/mysql/V28__alter_message_add_to_column.sql new file mode 100755 index 00000000..7b4b69d7 --- /dev/null +++ b/broker/migrate/mysql/V28__alter_message_add_to_column.sql @@ -0,0 +1,38 @@ +alter table t_messages add column `_to` varchar(64) DEFAULT NULL; + +alter table t_messages_0 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_1 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_2 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_3 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_4 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_5 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_6 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_7 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_8 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_9 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_10 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_11 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_12 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_13 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_14 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_15 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_16 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_17 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_18 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_19 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_20 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_21 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_22 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_23 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_24 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_25 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_26 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_27 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_28 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_29 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_30 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_31 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_32 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_33 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_34 add column `_to` varchar(64) DEFAULT NULL; +alter table t_messages_35 add column `_to` varchar(64) DEFAULT NULL; diff --git a/broker/src/main/java/io/moquette/imhandler/LoadRemoteMessagesHandler.java b/broker/src/main/java/io/moquette/imhandler/LoadRemoteMessagesHandler.java index 58ac09bd..6d3a5314 100644 --- a/broker/src/main/java/io/moquette/imhandler/LoadRemoteMessagesHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/LoadRemoteMessagesHandler.java @@ -32,6 +32,12 @@ public class LoadRemoteMessagesHandler extends IMHandler loadRemoteMessages(String user, WFCMessage.Conversation conversation, long beforeUid, int count) { - List messages = loadRemoteMessagesFromTable(user, conversation, beforeUid, count, MessageShardingUtil.getMessageTable(beforeUid)); - if (messages != null && messages.size() < count) { - String nexTable = MessageShardingUtil.getPreviousMessageTable(beforeUid); - if (!StringUtil.isNullOrEmpty(nexTable)) { - List nextMessages = loadRemoteMessagesFromTable(user, conversation, beforeUid, count - messages.size(), nexTable); - if (nextMessages != null) { - messages.addAll(nextMessages); + List messages = new ArrayList<>(); + long[] before = new long[1]; + before[0] = beforeUid; + boolean hasMore = loadRemoteMessagesFromTable(user, conversation, before, count, MessageShardingUtil.getMessageTable(beforeUid), messages); + if (messages.size() < count) { + if (hasMore) { + loadRemoteMessagesFromTable(user, conversation, before, count - messages.size(), MessageShardingUtil.getMessageTable(beforeUid), messages); + } else { + String nexTable = MessageShardingUtil.getPreviousMessageTable(before[0]); + if (!StringUtil.isNullOrEmpty(nexTable)) { + loadRemoteMessagesFromTable(user, conversation, before, count - messages.size(), nexTable, messages); } } } + return messages; } - List loadRemoteMessagesFromTable(String user, WFCMessage.Conversation conversation, long beforeUid, int count, String table) { - String sql = "select `_mid`, `_from`, `_type`, `_target`, `_line`, `_data`, `_dt` from " + table +" where"; + boolean loadRemoteMessagesFromTable(String user, WFCMessage.Conversation conversation, long[] before, int count, String table, List messages) { + 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 { @@ -747,7 +760,6 @@ public class DatabaseStore { Connection connection = null; PreparedStatement statement = null; ResultSet resultSet = null; - List out = new ArrayList<>(); try { connection = DBUtil.getConnection(); statement = connection.prepareStatement(sql); @@ -764,9 +776,12 @@ public class DatabaseStore { statement.setInt(index++, count); resultSet = statement.executeQuery(); while (resultSet.next()) { + count--; WFCMessage.Message.Builder builder = WFCMessage.Message.newBuilder(); index = 1; builder.setMessageId(resultSet.getLong(index++)); + before[0] = builder.getMessageId(); + builder.setFromUser(resultSet.getString(index++)); WFCMessage.Conversation.Builder cb = WFCMessage.Conversation.newBuilder(); cb.setType(resultSet.getInt(index++)); @@ -777,9 +792,22 @@ public class DatabaseStore { WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.parseFrom(blob.getBinaryStream()); builder.setContent(messageContent); + builder.setServerTimestamp(resultSet.getTimestamp(index++).getTime()); + String to = resultSet.getString(index++); + if (!StringUtil.isNullOrEmpty(to)) { + if (to.equals(user) || builder.getFromUser().equals(user)) { + builder.setToUser(to); + } else { + continue; + } + } WFCMessage.Message message = builder.build(); - out.add(message); + messages.add(message); + } + + if (count == 0) { + return true; } } catch (SQLException | IOException e) { e.printStackTrace(); @@ -787,7 +815,7 @@ public class DatabaseStore { } finally { DBUtil.closeDB(connection, statement, resultSet); } - return out; + return false; } void persistUserMessage(final String userId, final long messageId, final long messageSeq) { diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 79326b57..542e5625 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -2681,6 +2681,16 @@ public class MemoryMessagesStore implements IMessagesStore { return true; } + @Override + public Collection getChannelSubscriber(String channelId) { + MultiMap channelMembers = m_Server.getHazelcastInstance().getMultiMap(CHANNEL_LISTENERS); + if (channelMembers == null) { + return new ArrayList<>(); + } + + return channelMembers.get(channelId); + } + @Override public Set handleSensitiveWord(String message) { updateSensitiveWord(); diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index eb02f622..606e54e8 100755 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -196,6 +196,7 @@ public interface IMessagesStore { ErrorCode listenChannel(String operator, String channelId, boolean listen); WFCMessage.ChannelInfo getChannelInfo(String channelId); boolean checkUserInChannel(String user, String channelId); + Collection getChannelSubscriber(String channelId); Set handleSensitiveWord(String message); boolean addSensitiveWords(List words);