diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java new file mode 100644 index 00000000..c89aefb1 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java @@ -0,0 +1,77 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.MessageToByteEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; + +/** + * + * @author andrea + */ +public class MQTTEncoder extends MessageToByteEncoder { + + @Override + protected void encode(ChannelHandlerContext chc, AbstractMessage msg, ByteBuf bb) throws Exception { + if (msg.getMessageType() == AbstractMessage.CONNECT) { + encodeConnect(chc, (ConnectMessage)msg, bb); + } + } + + private void encodeConnect(ChannelHandlerContext chc, ConnectMessage message, ByteBuf out) { + ByteBuf staticHeaderBuff = chc.alloc().buffer(12); + staticHeaderBuff.writeBytes(Utils.encodeString("MQIsdp")); + + //version + staticHeaderBuff.writeByte(0x03); + + //connection flags and Strings + byte connectionFlags = 0; + if (message.isCleanSession()) { + connectionFlags |= 0x02; + } + if (message.isWillFlag()) { + connectionFlags |= 0x04; + } + connectionFlags |= ((message.getWillQos() & 0x03) << 3); + if (message.isWillRetain()) { + connectionFlags |= 0x020; + } + if (message.isPasswordFlag()) { + connectionFlags |= 0x040; + } + if (message.isUserFlag()) { + connectionFlags |= 0x080; + } + staticHeaderBuff.writeByte(connectionFlags); + + //Keep alive timer + staticHeaderBuff.writeShort(message.getKeepAlive()); + + //Variable part + ByteBuf variableHeaderBuff = chc.alloc().buffer(12); + if (message.getClientID() != null) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getClientID())); + if (message.isWillFlag()) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillTopic())); + variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillMessage())); + } + if (message.isUserFlag() && message.getUsername() != null) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getUsername())); + if (message.isPasswordFlag() && message.getPassword() != null) { + variableHeaderBuff.writeBytes(Utils.encodeString(message.getPassword())); + } + } + } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + ByteBuf buff = chc.alloc().buffer(); + buff.writeByte(AbstractMessage.CONNECT << 4); + buff.writeBytes(Utils.encodeRemainingLength(12 + variableHeaderSize)); + buff.writeBytes(staticHeaderBuff).writeBytes(variableHeaderBuff); + + out.writeBytes(buff); + } + +} 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 index 04267ff2..f0814bb3 100644 --- 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 @@ -2,6 +2,7 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.handler.codec.CorruptedFrameException; import java.io.UnsupportedEncodingException; import org.slf4j.LoggerFactory; @@ -11,6 +12,8 @@ import org.slf4j.LoggerFactory; */ public class Utils { + public static final int MAX_LENGTH_LIMIT = 268435455; + static byte readMessageType(ByteBuf in) { byte h1 = in.readByte(); byte messageType = (byte) ((h1 & 0x00F0) >> 4); @@ -60,6 +63,32 @@ public class Utils { return value; } + /** + * Encode the value in the format defined in specification as variable length + * array. + * + * @throws IllegalArgumentException if the value is not in the specification bounds + * [0..268435455]. + */ + static ByteBuf encodeRemainingLength(int value) throws CorruptedFrameException { + if (value > MAX_LENGTH_LIMIT || value < 0) { + throw new CorruptedFrameException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value); + } + + ByteBuf encoded = Unpooled.buffer(4); + byte digit; + do { + digit = (byte) (value % 128); + value = value / 128; + // if there are more digits to encode, set the top bit of this digit + if (value > 0) { + digit = (byte) (digit | 0x80); + } + encoded.writeByte(digit); + } while (value > 0); + return encoded; + } + /** * Load a string from the given buffer, reading first the two bytes of len * and then the UTF-8 bytes of the string. diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java new file mode 100644 index 00000000..6fe50536 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java @@ -0,0 +1,133 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.Unpooled; +import io.netty.buffer.UnpooledByteBufAllocator; +import io.netty.channel.ChannelHandlerContext; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.junit.Test; +import org.junit.Before; +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.*; +import static org.dna.mqtt.moquette.parser.netty.TestUtils.*; + + +/** + * + * @author andrea + */ +public class MQTTEncoderTest { + MQTTEncoder m_encoder = new MQTTEncoder(); + ChannelHandlerContext m_mockedContext; + + @Before + public void setUp() { + //mock the ChannelHandlerContext to return an UnpooledAllocator + m_mockedContext = mock(ChannelHandlerContext.class); + ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT; + when(m_mockedContext.alloc()).thenReturn(allocator); + } + + @Test + public void testBaseHeader() throws Exception { + ConnectMessage msg = new ConnectMessage(); + msg.setWillRetain(true); + msg.setWillQos((byte)2); + msg.setWillFlag(false); + msg.setCleanSession(true); + msg.setKeepAlive(512); + //mock the ChannelHandlerContext to return an UnpooledAllocator + /*ChannelHandlerContext mockedContext = mock(ChannelHandlerContext.class); + ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT; + when(mockedContext.alloc()).thenReturn(allocator);*/ + ByteBuf out = Unpooled.buffer(); + + //Exercise + m_encoder.encode(m_mockedContext, msg, out); + + //Verify + assertEquals(0x10, out.readByte()); //1 byte + assertEquals(12, out.readByte()); //remaining length + verifyString("MQIsdp", out); + assertEquals(0x03, out.readByte()); //protocol version + assertEquals(0x32, out.readByte()); //flags + assertEquals(2, out.readByte()); //keepAliveTimer msb + assertEquals(0, out.readByte()); //keepAliveTimer lsb + } + + + @Test + public void testCompleteHeader() throws Exception { + ConnectMessage msg = new ConnectMessage(); + msg.setWillRetain(true); + msg.setWillQos((byte)2); + msg.setWillFlag(true); + msg.setCleanSession(true); + msg.setKeepAlive(512); + + //variable part + msg.setClientID("ABCDEF"); + msg.setWillTopic("Topic"); + msg.setWillMessage("Message"); + + ByteBuf out = Unpooled.buffer(); + + //Exercise + m_encoder.encode(m_mockedContext, msg, out); + + //Verify + assertEquals(0x10, out.readByte()); //1 byte + assertEquals(36, out.readByte()); //remaining length + verifyString("MQIsdp", out); + assertEquals(0x03, out.readByte()); //protocol version + assertEquals(0x36, out.readByte()); //flags + assertEquals(2, out.readByte()); //keepAliveTimer msb + assertEquals(0, out.readByte()); //keepAliveTimer lsb + + //Variable part + verifyString("ABCDEF", out); + verifyString("Topic", out); + verifyString("Message", out); + } + + @Test + public void testCompleteHeaderWIthUser_password() throws Exception { + ConnectMessage msg = new ConnectMessage(); + msg.setUserFlag(true); + msg.setPasswordFlag(true); + msg.setWillRetain(true); + msg.setWillQos((byte)2); + msg.setWillFlag(true); + msg.setCleanSession(true); + msg.setKeepAlive(512); + + //variable part + msg.setClientID("ABCDEF"); + msg.setWillTopic("Topic"); + msg.setWillMessage("Message"); + msg.setUsername("Pablo"); + msg.setPassword("PBL"); + + ByteBuf out = Unpooled.buffer(); + + //Exercise + m_encoder.encode(m_mockedContext, msg, out); + + //Verify + assertEquals(0x10, out.readByte()); //1 byte + assertEquals(48, out.readByte()); //remaining length + verifyString("MQIsdp", out); + assertEquals(0x03, out.readByte()); //protocol version + assertEquals((byte)0xF6, (byte)out.readByte()); //flags + assertEquals(2, out.readByte()); //keepAliveTimer msb + assertEquals(0, out.readByte()); //keepAliveTimer lsb + + //Variable part + verifyString("ABCDEF", out); + verifyString("Topic", out); + verifyString("Message", out); + verifyString("Pablo", out);//username + verifyString("PBL", out);//password + } +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java new file mode 100644 index 00000000..d1d80e57 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java @@ -0,0 +1,49 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.io.UnsupportedEncodingException; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * Common test methods + * + * @author andrea + */ +public class TestUtils { + + /** + * Verify that the given bytes buffer of the given numBytes length is present + * in the buff starting from the current position. + */ + static void verifyBuff(int numBytes, byte[] bytes, ByteBuf buff) { + assertTrue(numBytes <= buff.readableBytes()); + byte[] toCheck = new byte[numBytes]; + buff.readBytes(toCheck); + + for (int i = 0; i < numBytes; i++) { + assertEquals(bytes[i], toCheck[i]); + } + } + + /** + * Verify the presence of the given string starting from the current position + * inside the buffer. + */ + static void verifyString(String str, ByteBuf buff) throws UnsupportedEncodingException { + ByteBuf tmpBuff = Unpooled.buffer(2); + byte[] raw = str.getBytes("UTF-8"); + tmpBuff.writeShort(raw.length); + tmpBuff.writeBytes(raw); + int buffLen = raw.length + 2; + verifyByteBuf(tmpBuff, buff.slice(buff.readerIndex(), buffLen)); + buff.skipBytes(buffLen); + } + + + static void verifyByteBuf(ByteBuf expected, ByteBuf found) { + assertEquals(expected, found); + } + +} 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 index 623f5171..adfbc366 100644 --- 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 @@ -2,9 +2,11 @@ package org.dna.mqtt.moquette.parser.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; -import static org.junit.Assert.assertEquals; +import io.netty.handler.codec.CorruptedFrameException; import org.junit.Before; import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.parser.netty.TestUtils.*; /** * @@ -47,4 +49,33 @@ public class UtilsTest { m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, (byte)0x7F}); assertEquals(268435455, Utils.decodeRemainingLenght(m_buff)); } + + @Test(expected=CorruptedFrameException.class) + public void testEncodeRemainingLength_invalid_upper() { + Utils.encodeRemainingLength(Utils.MAX_LENGTH_LIMIT + 1); + } + + @Test(expected=CorruptedFrameException.class) + public void testEncodeRemainingLength_invalid_lower() { + Utils.encodeRemainingLength(-1); + } + + @Test + public void testEncodeRemainingLenght() { + //1 byte length + verifyBuff(1, new byte[]{0}, Utils.encodeRemainingLength(0)); + verifyBuff(1, new byte[]{0x7F}, Utils.encodeRemainingLength(127)); + + //2 byte length + verifyBuff(2, new byte[]{(byte)0x80, 0x01}, Utils.encodeRemainingLength(128)); + verifyBuff(2, new byte[]{(byte)0xFF, 0x7F}, Utils.encodeRemainingLength(16383)); + + //3 byte length + verifyBuff(3, new byte[]{(byte)0x80, (byte)0x80, 0x01}, Utils.encodeRemainingLength(16384)); + verifyBuff(3, new byte[]{(byte)0xFF, (byte)0xFF, 0x7F}, Utils.encodeRemainingLength(2097151)); + + //4 byte length + verifyBuff(4, new byte[]{(byte)0x80, (byte)0x80, (byte)0x80, 0x01}, Utils.encodeRemainingLength(2097152)); + verifyBuff(4, new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, 0x7F}, Utils.encodeRemainingLength(268435455)); + } }