添加敏感词命中消息表t_sensitive_messages

broker/migrate/h2/V9__add_sensitive_messages.sql

@@ -0,0 +1,15 @@
+
+DROP TABLE IF EXISTS `t_sensitive_messages`;
+CREATE TABLE `t_sensitive_messages` (
+  `id` int(11) NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  `_mid` bigint(20) NOT NULL,
+  `_from` varchar(64) NOT NULL,
+  `_type` tinyint NOT NULL DEFAULT 0,
+  `_target` varchar(64) NOT NULL,
+  `_line` int(11) NOT NULL DEFAULT 0,
+  `_data` BLOB NOT NULL,
+  `_searchable_key` TEXT DEFAULT NULL,
+  `_dt` DATETIME NOT NULL DEFAULT NOW(),
+  `_content_type` int(11) NOT NULL DEFAULT '0',
+  UNIQUE INDEX `message_uid_index` (`_mid` ASC)
+);

broker/migrate/mysql/V9__add_sensitive_messages.sql

@@ -0,0 +1,18 @@
+
+DROP TABLE IF EXISTS `t_sensitive_messages`;
+CREATE TABLE `t_sensitive_messages` (
+    `id` int(11) NOT NULL PRIMARY KEY AUTO_INCREMENT,
+    `_mid` bigint(20) NOT NULL,
+    `_from` varchar(64) NOT NULL,
+    `_type` tinyint NOT NULL DEFAULT 0,
+    `_target` varchar(64) NOT NULL,
+    `_line` int(11) NOT NULL DEFAULT 0,
+    `_data` BLOB NOT NULL,
+    `_searchable_key` TEXT DEFAULT NULL,
+    `_dt` DATETIME NOT NULL,
+    `_content_type` int(11) NOT NULL DEFAULT '0',
+    INDEX `message_uid_index` (`_mid` DESC)
+  )
+ENGINE = InnoDB
+DEFAULT CHARACTER SET = utf8mb4
+COLLATE = utf8mb4_unicode_ci;

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

@@ -81,6 +81,7 @@ public class SendMessageHandler extends IMHandler<WFCMessage.Message> {
             if (!isAdmin && message.getContent().getType() == Text) {
                 Set<String> matched = m_messagesStore.handleSensitiveWord(message.getContent().getSearchableContent());
                 if (matched != null && !matched.isEmpty()) {
+                    m_messagesStore.storeSensitiveMessage(message);
                     if (mSensitiveType == 0) {
                         errorCode = ErrorCode.ERROR_CODE_SENSITIVE_MATCHED;
                     } else if(mSensitiveType == 1) {

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

@@ -424,6 +424,57 @@ public class DatabaseStore {
         });
     }
 
+    void persistSensitiveMessage(final WFCMessage.Message message) {
+        if(message.getContent().getPersistFlag() == Transparent) {
+            return;
+        }
+
+        mScheduler.execute(()-> {
+            Connection connection = null;
+            PreparedStatement statement = null;
+            try {
+                connection = DBUtil.getConnection();
+                String sql = "insert into " + "t_sensitive_messages" +
+                    " (`_mid`, `_from`, `_type`, `_target`, `_line`, `_data`, `_searchable_key`, `_dt`, `_content_type`) values(?, ?, ?, ?, ?, ?, ?, ?, ?)" +
+                    " ON DUPLICATE KEY UPDATE " +
+                    "`_data` = ?," +
+                    "`_searchable_key` = ?," +
+                    "`_dt` = ?," +
+                    "`_content_type` = ?";
+
+                String searchableContent = message.getContent().getSearchableContent() == null ? "" : message.getContent().getSearchableContent();
+
+                statement = connection.prepareStatement(sql);
+                int index = 1;
+                statement.setLong(index++, message.getMessageId());
+                statement.setString(index++, message.getFromUser());
+                statement.setInt(index++, message.getConversation().getType());
+                statement.setString(index++, message.getConversation().getTarget());
+                statement.setInt(index++, message.getConversation().getLine());
+                Blob blob = connection.createBlob();
+                blob.setBytes(1, message.getContent().toByteArray());
+                statement.setBlob(index++, blob);
+                statement.setString(index++, searchableContent);
+                statement.setTimestamp(index++, new Timestamp(message.getServerTimestamp()));
+                statement.setInt(index++, message.getContent().getType());
+
+                statement.setBlob(index++, blob);
+                statement.setString(index++, searchableContent);
+                statement.setTimestamp(index++, new Timestamp(message.getServerTimestamp()));
+                statement.setInt(index++, message.getContent().getType());
+
+                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);
+            }
+        });
+    }
+
     Map<Long, MessageBundle> getMessages(Collection<Long> keys) {
         Map<String, List<Long>> messageTableMap = new HashMap<>();
         for (Long key : keys) {

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

@@ -169,6 +169,12 @@ public class MemoryMessagesStore implements IMessagesStore {
         return message;
     }
 
+    @Override
+    public void storeSensitiveMessage(WFCMessage.Message message) {
+        databaseStore.persistSensitiveMessage(message);
+    }
+
+
     @Override
     public int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder messageBuilder, Set<String> notifyReceivers) {
         WFCMessage.Message message = messageBuilder.build();

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

@@ -96,6 +96,7 @@ public interface IMessagesStore {
 
     DatabaseStore getDatabaseStore();
     WFCMessage.Message storeMessage(String fromUser, String fromClientId, WFCMessage.Message message);
+    void storeSensitiveMessage(WFCMessage.Message message);
 	int getNotifyReceivers(String fromUser, WFCMessage.Message.Builder message, Set<String> notifyReceivers);
     Set<String> getAllEnds();
     WFCMessage.PullMessageResult fetchMessage(String user, String exceptClientId, long fromMessageId, int pullType);