From 8e8a2952008b11f2b95c507f867ee90572ac3a34 Mon Sep 17 00:00:00 2001 From: Andreas Mager Date: Tue, 25 Apr 2017 17:25:37 +0200 Subject: [PATCH] broker/BrokerInterceptor: ByteBuf.retain & release added broker/NettyMQTTHandler: ByteBuf.release added broker/HazelcastListener: release added broker/pom: leak detection enabled for surefire tests broker: ReferenceCountUtil.release used --- broker/pom.xml | 3 +++ .../io/moquette/server/HazelcastListener.java | 6 ++++- .../server/netty/NettyMQTTHandler.java | 3 +++ .../moquette/spi/impl/BrokerInterceptor.java | 23 +++++++++++++------ 4 files changed, 27 insertions(+), 8 deletions(-) diff --git a/broker/pom.xml b/broker/pom.xml index dfc560e1..8de1d653 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -190,6 +190,9 @@ org.apache.maven.plugins maven-surefire-plugin + + PARANOID + **/*Test.java diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java index 0fc943fc..2db2fa0a 100644 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ b/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 { @Override public void onMessage(Message 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 { 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); } } } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 52c170ad..28419b8d 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/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); } } diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 61d0ca38..c59ad4d2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/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(); - String topic = msg.variableHeader().topicName(); - for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) { - LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}", - clientID, messageId, topic, handler.getID()); - executor.execute(() -> handler.onPublish(new InterceptPublishMessage(msg, clientID, username))); - } + msg.retain(); + + executor.execute(() -> { + try { + int messageId = msg.variableHeader().messageId(); + 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