diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index 1036c84d..9c07b14a 100755 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/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 userMaxPullSeq = new ConcurrentHashMap<>(); - private SensitiveFilter mSensitiveFilter; + private AtomicReference mSensitiveFilter; private volatile long lastUpdateSensitiveTime = 0; private ConcurrentHashMap userGlobalSlientMap = new ConcurrentHashMap<>(); @@ -145,7 +146,7 @@ public class MemoryMessagesStore implements IMessagesStore { } } Set sensitiveWords = databaseStore.getSensitiveWord(); - mSensitiveFilter = new SensitiveFilter(sensitiveWords); + mSensitiveFilter.set(new SensitiveFilter(sensitiveWords)); } } @@ -2492,7 +2493,7 @@ public class MemoryMessagesStore implements IMessagesStore { @Override public Set handleSensitiveWord(String message) { updateSensitiveWord(); - return mSensitiveFilter.getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH); + return mSensitiveFilter.get().getSensitiveWords(message, SensitiveFilter.MatchType.MAX_MATCH); } @Override diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 19806b42..5723d466 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/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; }