Merge pull request #124 from k81/wildfirechat

Fix race condition during SensitiveFilter update and potential resource leak of GZIPOutputStream on exception

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

@@ -41,6 +41,7 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
@@ -110,7 +111,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     private final DatabaseStore databaseStore;
     private ConcurrentHashMap<String, Long> userMaxPullSeq = new ConcurrentHashMap<>();
 
-    private SensitiveFilter mSensitiveFilter;
+    private AtomicReference<SensitiveFilter> mSensitiveFilter;
     private volatile long lastUpdateSensitiveTime = 0;
 
     private ConcurrentHashMap<String, Boolean> userGlobalSlientMap = new ConcurrentHashMap<>();
@@ -145,7 +146,7 @@ public class MemoryMessagesStore implements IMessagesStore {
                 }
             }
             Set<String> sensitiveWords = databaseStore.getSensitiveWord();
-            mSensitiveFilter = new SensitiveFilter(sensitiveWords);
+            mSensitiveFilter.set(new SensitiveFilter(sensitiveWords));
         }
     }
 
@@ -2492,7 +2493,7 @@ public class MemoryMessagesStore implements IMessagesStore {
     @Override
     public Set<String> handleSensitiveWord(String message) {
         updateSensitiveWord();
-        return mSensitiveFilter.getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH);
+        return mSensitiveFilter.get().getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH);
     }
 
     @Override

broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java

@@ -179,16 +179,15 @@ public class Qos1PublishHandler extends QosPublishHandler {
                             MemorySessionStore.Session session = m_sessionStore.getSession(clientID);
                             if (session != null && session.getUsername().equals(fromUser)) {
                                 if (data.length > 7*1024 && session.getMqttVersion().protocolLevel() >= MqttVersion.Wildfire_1.protocolLevel()) {
-                                    ByteArrayOutputStream out = new ByteArrayOutputStream();
-                                    GZIPOutputStream gzip;
-                                    try {
-                                        gzip = new GZIPOutputStream(out);
+                                    try (
+                                        ByteArrayOutputStream out = new ByteArrayOutputStream();
+                                        GZIPOutputStream gzip = new GZIPOutputStream(out)
+                                    ) {
                                         gzip.write(data);
-                                        gzip.close();
+                                        data = out.toByteArray();
                                     } catch (Exception e) {
                                         e.printStackTrace();
                                     }
-                                    data = out.toByteArray();
                                     code = (byte)ErrorCode.ERROR_CODE_SUCCESS_GZIPED.code;
                                 }