diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java new file mode 100644 index 00000000..a21b1f53 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java @@ -0,0 +1,146 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.CorruptedFrameException; +import java.io.UnsupportedEncodingException; +import java.util.List; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; + +/** + * + * @author andrea + */ +public class ConnectDecoder extends DemuxDecoder { + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws UnsupportedEncodingException { + in.resetReaderIndex(); + //Common decoding part + ConnectMessage message = new ConnectMessage(); + if (!decodeCommonHeader(message, in)) { + in.resetReaderIndex(); + return; + } + int remainingLength = message.getRemainingLength(); + int start = in.readerIndex(); + + //Connect specific decoding part + //ProtocolName 8 bytes + if (in.readableBytes() < 12) { + in.resetReaderIndex(); + return; + } + byte[] encProtoName = new byte[6]; + in.skipBytes(2); //size, is 0x06 + in.readBytes(encProtoName); + String protoName = new String(encProtoName, "UTF-8"); + if (!"MQIsdp".equals(protoName)) { + in.resetReaderIndex(); + throw new CorruptedFrameException("Invalid protoName: " + protoName); + } + message.setProtocolName(protoName); + + //ProtocolVersion 1 byte (value 0x03) + message.setProcotolVersion(in.readByte()); + + //Connection flag + byte connFlags = in.readByte(); + boolean cleanSession = ((connFlags & 0x02) >> 1) == 1 ? true : false; + boolean willFlag = ((connFlags & 0x04) >> 2) == 1 ? true : false; + byte willQos = (byte) ((connFlags & 0x18) >> 3); + if (willQos > 2) { + in.resetReaderIndex(); + throw new CorruptedFrameException("Expected will QoS in range 0..2 but found: " + willQos); + } + boolean willRetain = ((connFlags & 0x20) >> 5) == 1 ? true : false; + boolean passwordFlag = ((connFlags & 0x40) >> 6) == 1 ? true : false; + boolean userFlag = ((connFlags & 0x80) >> 7) == 1 ? true : false; + //a password is true iff user is true. + if (!userFlag && passwordFlag) { + in.resetReaderIndex(); + throw new CorruptedFrameException("Expected password flag to true if the user flag is true but was: " + passwordFlag); + } + message.setCleanSession(cleanSession); + message.setWillFlag(willFlag); + message.setWillQos(willQos); + message.setWillRetain(willRetain); + message.setPasswordFlag(passwordFlag); + message.setUserFlag(userFlag); + + //Keep Alive timer 2 bytes + //int keepAlive = Utils.readWord(in); + int keepAlive = in.readShort(); + message.setKeepAlive(keepAlive); + + if (remainingLength == 12) { + out.add(message); + return; + } + + //Decode the ClientID + String clientID = Utils.decodeString(in); + if (clientID == null) { + in.resetReaderIndex(); + return; + } + message.setClientID(clientID); + + //Decode willTopic + if (willFlag) { + String willTopic = Utils.decodeString(in); + if (willTopic == null) { + in.resetReaderIndex(); + return; + } + message.setWillTopic(willTopic); + } + + //Decode willMessage + if (willFlag) { + String willMessage = Utils.decodeString(in); + if (willMessage == null) { + in.resetReaderIndex(); + return; + } + message.setWillMessage(willMessage); + } + + //Compatibility check wieth v3.0, remaining length has precedence over + //the user and password flags + int readed = in.readerIndex() - start; + if (readed == remainingLength) { + out.add(message); + return; + } + + //Decode username + if (userFlag) { + String userName = Utils.decodeString(in); + if (userName == null) { + in.resetReaderIndex(); + return; + } + message.setUsername(userName); + } + + readed = in.readerIndex() - start; + if (readed == remainingLength) { + out.add(message); + return; + } + + //Decode password + if (passwordFlag) { + String password = Utils.decodeString(in); + if (password == null) { + in.resetReaderIndex(); + return; + } + message.setPassword(password); + } + + out.add(message); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxDecoder.java new file mode 100644 index 00000000..8c055fb7 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/DemuxDecoder.java @@ -0,0 +1,38 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import java.util.List; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; + +/** + * + * @author andrea + */ +abstract class DemuxDecoder { + abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception; + + protected boolean decodeCommonHeader(AbstractMessage message, ByteBuf in) { + //Common decoding part + if (in.readableBytes() < 2) { + return false; + } + byte h1 = in.readByte(); + byte messageType = (byte) ((h1 & 0x00F0) >> 4); + boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1); + byte qosLevel = (byte) ((h1 & 0x0006) >> 1); + boolean retainFlag = ((byte) (h1 & 0x0001) == 1); + int remainingLength = Utils.decodeRemainingLenght(in); + if (remainingLength == -1) { + return false; + } + + message.setMessageType(messageType); + message.setDupFlag(dupFlag); + message.setQos(AbstractMessage.QOSType.values()[qosLevel]); + message.setRetainFlag(retainFlag); + message.setRemainingLength(remainingLength); + return true; + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java index 708afa66..3021a397 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java @@ -4,15 +4,22 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.handler.codec.CorruptedFrameException; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; -import org.dna.mqtt.moquette.proto.messages.ConnectMessage; /** * * @author andrea */ public class MQTTDecoder extends ByteToMessageDecoder { + + private Map m_decoderMap = new HashMap(); + + MQTTDecoder() { + m_decoderMap.put(AbstractMessage.CONNECT, new ConnectDecoder()); + } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { @@ -25,159 +32,10 @@ public class MQTTDecoder extends ByteToMessageDecoder { byte messageType = Utils.readMessageType(in); - if (messageType == AbstractMessage.CONNECT) { - in.resetReaderIndex(); - //Common decoding part - ConnectMessage message = new ConnectMessage(); - if (!decodeCommonHeader(message, in)) { - in.resetReaderIndex(); - return; - } - int remainingLength = message.getRemainingLength(); - int start = in.readerIndex(); - - //Connect specific decoding part - //ProtocolName 8 bytes - if (in.readableBytes() < 12) { - in.resetReaderIndex(); - return; - } - byte[] encProtoName = new byte[6]; - in.skipBytes(2); //size, is 0x06 - in.readBytes(encProtoName); - String protoName = new String(encProtoName, "UTF-8"); - if (!"MQIsdp".equals(protoName)) { - in.resetReaderIndex(); - throw new CorruptedFrameException("Invalid protoName: " + protoName); - } - message.setProtocolName(protoName); - - //ProtocolVersion 1 byte (value 0x03) - message.setProcotolVersion(in.readByte()); - - //Connection flag - byte connFlags = in.readByte(); - boolean cleanSession = ((connFlags & 0x02) >> 1) == 1 ? true : false; - boolean willFlag = ((connFlags & 0x04) >> 2) == 1 ? true : false; - byte willQos = (byte) ((connFlags & 0x18) >> 3); - if (willQos > 2) { - in.resetReaderIndex(); - throw new CorruptedFrameException("Expected will QoS in range 0..2 but found: " + willQos); - } - boolean willRetain = ((connFlags & 0x20) >> 5) == 1 ? true : false; - boolean passwordFlag = ((connFlags & 0x40) >> 6) == 1 ? true : false; - boolean userFlag = ((connFlags & 0x80) >> 7) == 1 ? true : false; - //a password is true iff user is true. - if (!userFlag && passwordFlag) { - in.resetReaderIndex(); - throw new CorruptedFrameException("Expected password flag to true if the user flag is true but was: " + passwordFlag); - } - message.setCleanSession(cleanSession); - message.setWillFlag(willFlag); - message.setWillQos(willQos); - message.setWillRetain(willRetain); - message.setPasswordFlag(passwordFlag); - message.setUserFlag(userFlag); - - //Keep Alive timer 2 bytes - //int keepAlive = Utils.readWord(in); - int keepAlive = in.readShort(); - message.setKeepAlive(keepAlive); - - if (remainingLength == 12) { - out.add(message); - return; - } - - //Decode the ClientID - String clientID = Utils.decodeString(in); - if (clientID == null) { - in.resetReaderIndex(); - return; - } - message.setClientID(clientID); - - //Decode willTopic - if (willFlag) { - String willTopic = Utils.decodeString(in); - if (willTopic == null) { - in.resetReaderIndex(); - return; - } - message.setWillTopic(willTopic); - } - - //Decode willMessage - if (willFlag) { - String willMessage = Utils.decodeString(in); - if (willMessage == null) { - in.resetReaderIndex(); - return; - } - message.setWillMessage(willMessage); - } - - //Compatibility check wieth v3.0, remaining length has precedence over - //the user and password flags - int readed = in.readerIndex() - start; - if (readed == remainingLength) { - out.add(message); - return; - } - - //Decode username - if (userFlag) { - String userName = Utils.decodeString(in); - if (userName == null) { - in.resetReaderIndex(); - return; - } - message.setUsername(userName); - } - - readed = in.readerIndex() - start; - if (readed == remainingLength) { - out.add(message); - return; - } - - //Decode password - if (passwordFlag) { - String password = Utils.decodeString(in); - if (password == null) { - in.resetReaderIndex(); - return; - } - message.setPassword(password); - } - - out.add(message); - return; + DemuxDecoder decoder = m_decoderMap.get(messageType); + if (decoder == null) { + throw new CorruptedFrameException("Can't find any suitable decoder for message type: " + messageType); } + decoder.decode(ctx, in, out); } - - - protected boolean decodeCommonHeader(AbstractMessage message, ByteBuf in) { - //Common decoding part - if (in.readableBytes() < 2) { - return false; - } - byte h1 = in.readByte(); - byte messageType = (byte) ((h1 & 0x00F0) >> 4); - boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1); - byte qosLevel = (byte) ((h1 & 0x0006) >> 1); - boolean retainFlag = ((byte) (h1 & 0x0001) == 1); - int remainingLength = Utils.decodeRemainingLenght(in); - if (remainingLength == -1) { - return false; - } - - message.setMessageType(messageType); - message.setDupFlag(dupFlag); - message.setQos(AbstractMessage.QOSType.values()[qosLevel]); - message.setRetainFlag(retainFlag); - message.setRemainingLength(remainingLength); - return true; - } - } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java similarity index 97% rename from netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoderTest.java rename to netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java index 121fffaf..f457e99e 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoderTest.java @@ -16,14 +16,14 @@ import org.junit.Before; import org.junit.Test; -public class MQTTDecoderTest { +public class ConnectDecoderTest { ByteBuf m_buff; - MQTTDecoder m_msgdec; + ConnectDecoder m_msgdec; @Before public void setUp() { - m_msgdec = new MQTTDecoder(); + m_msgdec = new ConnectDecoder(); } @Test