diff --git a/broker/src/main/java/io/moquette/broker/NettyUtils.java b/broker/src/main/java/io/moquette/broker/NettyUtils.java index 1d024c1b..bbbb8cfa 100644 --- a/broker/src/main/java/io/moquette/broker/NettyUtils.java +++ b/broker/src/main/java/io/moquette/broker/NettyUtils.java @@ -16,8 +16,11 @@ package io.moquette.broker; +import java.io.IOException; + import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.mqtt.MqttMessage; import io.netty.util.Attribute; import io.netty.util.AttributeKey; @@ -69,6 +72,25 @@ public final class NettyUtils { return (String) channel.attr(NettyUtils.ATTR_KEY_USERNAME).get(); } - private NettyUtils() { + /** + * Validate that the provided message is an MqttMessage and that it does not contain a failed result. + * + * @param message to be validated + * @return the casted provided message + * @throws IOException in case of an fail message this will wrap the root cause + * @throws ClassCastException if the provided message is no MqttMessage + */ + public static MqttMessage validateMessage(Object message) throws IOException, ClassCastException { + MqttMessage msg = (MqttMessage) message; + if (msg.decoderResult() != null && msg.decoderResult().isFailure()) { + throw new IOException("invalid massage", msg.decoderResult().cause()); + } + if (msg.fixedHeader() == null) { + throw new IOException("Unknown packet, no fixedHeader present, no cause provided"); + } + return msg; + } + + private NettyUtils() { } } diff --git a/broker/src/main/java/io/moquette/broker/NewNettyMQTTHandler.java b/broker/src/main/java/io/moquette/broker/NewNettyMQTTHandler.java index 9935beb8..d5bb2d83 100644 --- a/broker/src/main/java/io/moquette/broker/NewNettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/broker/NewNettyMQTTHandler.java @@ -52,10 +52,7 @@ public class NewNettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception { - MqttMessage msg = (MqttMessage) message; - if (msg.fixedHeader() == null) { - throw new IOException("Unknown packet"); - } + MqttMessage msg = NettyUtils.validateMessage(message); final MQTTConnection mqttConnection = mqttConnection(ctx.channel()); try { mqttConnection.handleMessage(msg); diff --git a/broker/src/main/java/io/moquette/broker/metrics/MQTTMessageLogger.java b/broker/src/main/java/io/moquette/broker/metrics/MQTTMessageLogger.java index 198b0b6d..601a633a 100644 --- a/broker/src/main/java/io/moquette/broker/metrics/MQTTMessageLogger.java +++ b/broker/src/main/java/io/moquette/broker/metrics/MQTTMessageLogger.java @@ -58,10 +58,7 @@ public class MQTTMessageLogger extends ChannelDuplexHandler { if (!(message instanceof MqttMessage)) { return; } - MqttMessage msg = (MqttMessage) message; - if (msg.fixedHeader() == null) { - throw new IOException("Unknown packet"); - } + MqttMessage msg = NettyUtils.validateMessage(message); String clientID = NettyUtils.clientID(ctx.channel()); MqttMessageType messageType = msg.fixedHeader().messageType(); switch (messageType) {