Merge pull request #302 from AndreasMager/add-ByteBuf.release

Add ByteBuf.release

broker/pom.xml

@@ -190,6 +190,9 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
                 <configuration>
+                    <systemPropertyVariables>
+                        <io.netty.leakDetectionLevel>PARANOID</io.netty.leakDetectionLevel>
+                    </systemPropertyVariables>
                     <includes>
                         <include>**/*Test.java</include>
                     </includes>

broker/src/main/java/io/moquette/server/HazelcastListener.java

@@ -22,6 +22,7 @@ import io.moquette.interception.HazelcastMsg;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.mqtt.*;
+import io.netty.util.ReferenceCountUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,6 +38,7 @@ public class HazelcastListener implements MessageListener<HazelcastMsg> {
 
     @Override
     public void onMessage(Message<HazelcastMsg> msg) {
+        ByteBuf payload = null;
         try {
             if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) {
                 HazelcastMsg hzMsg = msg.getMessageObject();
@@ -47,12 +49,14 @@ public class HazelcastListener implements MessageListener<HazelcastMsg> {
                 MqttQoS qos = MqttQoS.valueOf(hzMsg.getQos());
                 MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
                 MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(hzMsg.getTopic(), 0);
-                ByteBuf payload = Unpooled.wrappedBuffer(hzMsg.getPayload());
+                payload = Unpooled.wrappedBuffer(hzMsg.getPayload());
                 MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, payload);
                 server.internalPublish(publishMessage, hzMsg.getClientId());
             }
         } catch (Exception ex) {
             LOG.error("error polling hazelcast msg queue", ex);
+        } finally {
+            ReferenceCountUtil.release(payload);
         }
     }
 }

broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java

@@ -21,6 +21,7 @@ import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.mqtt.*;
+import io.netty.util.ReferenceCountUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
@@ -87,6 +88,8 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
             LOG.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex);
             ctx.fireExceptionCaught(ex);
             ctx.close();
+        } finally {
+            ReferenceCountUtil.release(msg);
         }
     }
 

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

@@ -24,6 +24,7 @@ import io.moquette.server.config.IConfig;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.netty.handler.codec.mqtt.MqttConnectMessage;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import io.netty.util.ReferenceCountUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.util.HashMap;
@@ -118,13 +119,21 @@ final class BrokerInterceptor implements Interceptor {
 
     @Override
     public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) {
-        int messageId = msg.variableHeader().messageId();
+        msg.retain();
-        String topic = msg.variableHeader().topicName();
+
-        for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) {
+        executor.execute(() -> {
-            LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}",
+                try {
-                clientID, messageId, topic, handler.getID());
+                    int messageId = msg.variableHeader().messageId();
-            executor.execute(() -> handler.onPublish(new InterceptPublishMessage(msg, clientID, username)));
+                    String topic = msg.variableHeader().topicName();
-        }
+                    for (InterceptHandler handler : handlers.get(InterceptPublishMessage.class)) {
+                        LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, "
+                                + "interceptorId={}", clientID, messageId, topic, handler.getID());
+                        handler.onPublish(new InterceptPublishMessage(msg, clientID, username));
+                    }
+                } finally {
+                    ReferenceCountUtil.release(msg);
+                }
+        });
     }
 
     @Override