diff --git a/netty_parser/pom.xml b/netty_parser/pom.xml index 03628397..a33bafe4 100644 --- a/netty_parser/pom.xml +++ b/netty_parser/pom.xml @@ -19,6 +19,12 @@ + + org.dna.mqtt + moquette-parser-commons + 0.2 + + io.netty netty-all 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 new file mode 100644 index 00000000..708afa66 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java @@ -0,0 +1,183 @@ +package org.dna.mqtt.moquette.parser.netty; + +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.List; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; + +/** + * + * @author andrea + */ +public class MQTTDecoder extends ByteToMessageDecoder { + + @Override + protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + in.markReaderIndex(); + if (!Utils.checkHeaderAvailability(in)) { + in.resetReaderIndex(); + return; + } + in.resetReaderIndex(); + + 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; + } + } + + + 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/Utils.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java new file mode 100644 index 00000000..04267ff2 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java @@ -0,0 +1,105 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.io.UnsupportedEncodingException; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class Utils { + + static byte readMessageType(ByteBuf in) { + byte h1 = in.readByte(); + byte messageType = (byte) ((h1 & 0x00F0) >> 4); + return messageType; + } + + static boolean checkHeaderAvailability(ByteBuf in) { + if (in.readableBytes() < 1) { + return false; + } + //byte h1 = in.get(); + //byte messageType = (byte) ((h1 & 0x00F0) >> 4); + in.skipBytes(1); //skip the messageType byte + + int remainingLength = Utils.decodeRemainingLenght(in); + if (remainingLength == -1) { + return false; + } + + //check remaining length + if (in.readableBytes() < remainingLength) { + return false; + } + + //return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK; + return true; + } + + /** + * Decode the variable remaining length as defined in MQTT v3.1 specification + * (section 2.1). + * + * @return the decoded length or -1 if needed more data to decode the length field. + */ + static int decodeRemainingLenght(ByteBuf in) { + int multiplier = 1; + int value = 0; + byte digit; + do { + if (in.readableBytes() < 1) { + return -1; + } + digit = in.readByte(); + value += (digit & 0x7F) * multiplier; + multiplier *= 128; + } while ((digit & 0x80) != 0); + return value; + } + + /** + * Load a string from the given buffer, reading first the two bytes of len + * and then the UTF-8 bytes of the string. + * + * @return the decoded string or null if NEED_DATA + */ + static String decodeString(ByteBuf in) throws UnsupportedEncodingException { + if (in.readableBytes() < 2) { + return null; + } + //int strLen = Utils.readWord(in); + int strLen = in.readShort(); + if (in.readableBytes() < strLen) { + return null; + } + byte[] strRaw = new byte[strLen]; + in.readBytes(strRaw); + + return new String(strRaw, "UTF-8"); + } + + + /** + * Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded + * string content. + */ + static ByteBuf encodeString(String str) { + ByteBuf out = Unpooled.buffer(2); + byte[] raw; + try { + raw = str.getBytes("UTF-8"); + //NB every Java platform has got UTF-8 encoding by default, so this + //exception are never raised. + } catch (UnsupportedEncodingException ex) { + LoggerFactory.getLogger(Utils.class).error(null, ex); + return null; + } + //Utils.writeWord(out, raw.length); + out.writeShort(raw.length); + out.writeBytes(raw); + return out; + } +} 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/MQTTDecoderTest.java new file mode 100644 index 00000000..18e57be7 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoderTest.java @@ -0,0 +1,80 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.io.UnsupportedEncodingException; +import java.util.ArrayList; +import java.util.List; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import org.junit.Before; +import org.junit.Test; + + +public class MQTTDecoderTest { + + ByteBuf m_buff; + MQTTDecoder m_msgdec; + + @Before + public void setUp() { + m_msgdec = new MQTTDecoder(); + } + + @Test + public void testBaseHeader() throws Exception { + m_buff = Unpooled.buffer(14); + initBaseHeader(m_buff); + //m_buff.flip(); + List results = new ArrayList(); + + //Excercise + m_msgdec.decode(null, m_buff, results); + + //Verify + assertFalse(results.isEmpty()); + verifyBaseHeader((ConnectMessage)results.get(0)); + } + + + private void initBaseHeader(ByteBuf buff) throws UnsupportedEncodingException { + initHeader(buff, (byte)0x0C); + } + + private void initHeader(ByteBuf buff, byte remaingLength) throws UnsupportedEncodingException { + buff.clear().writeByte((byte)(AbstractMessage.CONNECT << 4)).writeByte(remaingLength); + //Proto name + encodeString(buff, "MQIsdp"); + //version + buff.writeByte(3); + //conn flags + buff.writeByte(0xCE); + //keepAlive + buff.writeBytes(new byte[]{(byte)0, (byte) 0x0A}); + } + + /** + * Encode and insert the given string into the given buff + */ + private ByteBuf encodeString(ByteBuf buff, String str) throws UnsupportedEncodingException { + buff.writeBytes(Utils.encodeString(str)); + return buff; + } + + private void verifyBaseHeader(ConnectMessage connMessage) { + assertNotNull(connMessage); + assertEquals("MQIsdp", connMessage.getProtocolName()); + assertEquals(3, connMessage.getProcotolVersion()); + assertTrue(connMessage.isUserFlag()); + assertTrue(connMessage.isPasswordFlag()); + assertTrue(connMessage.isCleanSession()); + assertEquals(10, connMessage.getKeepAlive()); + assertTrue(connMessage.isWillFlag()); + assertFalse(connMessage.isWillRetain()); + assertEquals(1, connMessage.getWillQos()); + } +} \ No newline at end of file diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UtilsTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UtilsTest.java new file mode 100644 index 00000000..623f5171 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UtilsTest.java @@ -0,0 +1,50 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import static org.junit.Assert.assertEquals; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author andrea + */ +/*@RunWith(Suite.class) +@Suite.SuiteClasses({})*/ +public class UtilsTest { + + ByteBuf m_buff; + + @Before + public void setUp() { + m_buff = Unpooled.buffer(4); + } + + @Test + public void testDecodeRemainingLength() { + //1 byte length + m_buff.writeByte(0x0); + assertEquals(0, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().writeByte(0x007F); + assertEquals(127, Utils.decodeRemainingLenght(m_buff)); + + //2 byte length + m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x01}); + assertEquals(128, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0x7F}); + assertEquals(16383, Utils.decodeRemainingLenght(m_buff)); + + //3 byte length + m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x80, (byte)0x01}); + assertEquals(16384, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0x7F}); + assertEquals(2097151, Utils.decodeRemainingLenght(m_buff)); + + //4 byte length + m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x80, (byte)0x80, (byte)0x01}); + assertEquals(2097152, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, (byte)0x7F}); + assertEquals(268435455, Utils.decodeRemainingLenght(m_buff)); + } +}