Fixed memory leaks in MqttTransportHandler (#1787)

@@ -34,6 +34,7 @@ import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
 import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.ssl.SslHandler;
+import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.GenericFutureListener;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
@@ -112,10 +113,14 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     @Override
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
         log.trace("[{}] Processing msg: {}", sessionId, msg);
         log.trace("[{}] Processing msg: {}", sessionId, msg);
-        if (msg instanceof MqttMessage) {
-            processMqttMsg(ctx, (MqttMessage) msg);
-        } else {
-            ctx.close();
+        try {
+            if (msg instanceof MqttMessage) {
+                processMqttMsg(ctx, (MqttMessage) msg);
+            } else {
+                ctx.close();
+            }
+        } finally {
+            ReferenceCountUtil.safeRelease(msg);
         }
         }
     }
     }
 
 
@@ -422,6 +427,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     }
     }
 
 
     @Override
     @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) {
+        ctx.flush();
+    }
+
+    @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         log.error("[{}] Unexpected Exception", sessionId, cause);
         log.error("[{}] Unexpected Exception", sessionId, cause);
         ctx.close();
         ctx.close();

@@ -212,18 +212,14 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
     }
     }
 
 
     private static String validatePayload(UUID sessionId, ByteBuf payloadData, boolean isEmptyPayloadAllowed) throws AdaptorException {
     private static String validatePayload(UUID sessionId, ByteBuf payloadData, boolean isEmptyPayloadAllowed) throws AdaptorException {
-        try {
-            String payload = payloadData.toString(UTF8);
-            if (payload == null) {
-                log.warn("[{}] Payload is empty!", sessionId);
-                if (!isEmptyPayloadAllowed) {
-                    throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
-                }
+        String payload = payloadData.toString(UTF8);
+        if (payload == null) {
+            log.warn("[{}] Payload is empty!", sessionId);
+            if (!isEmptyPayloadAllowed) {
+                throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
             }
             }
-            return payload;
-        } finally {
-            payloadData.release();
         }
         }
+        return payload;
     }
     }
 
 
 }
 }