From 28b37f68ea43df3f8f9096351d5020839d75908b Mon Sep 17 00:00:00 2001 From: andrea Date: Thu, 12 Sep 2013 12:38:31 +0200 Subject: [PATCH] Added encode/decode for Publish message --- .../moquette/parser/netty/MQTTDecoder.java | 1 + .../moquette/parser/netty/MQTTEncoder.java | 1 + .../moquette/parser/netty/PublishDecoder.java | 61 ++++++ .../moquette/parser/netty/PublishEncoder.java | 45 +++++ .../dna/mqtt/moquette/parser/netty/Utils.java | 25 +++ .../parser/netty/ConnectEncoderTest.java | 4 - .../parser/netty/PublishDecoderTest.java | 167 +++++++++++++++++ .../parser/netty/PublishEncoderTest.java | 175 ++++++++++++++++++ .../mqtt/moquette/parser/netty/TestUtils.java | 16 ++ .../mqtt/moquette/parser/netty/UtilsTest.java | 12 ++ 10 files changed, 503 insertions(+), 4 deletions(-) create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java create mode 100644 netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java create mode 100644 netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java 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 86fb02c5..8cfe6e9d 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 @@ -20,6 +20,7 @@ public class MQTTDecoder extends ByteToMessageDecoder { MQTTDecoder() { m_decoderMap.put(AbstractMessage.CONNECT, new ConnectDecoder()); m_decoderMap.put(AbstractMessage.CONNACK, new ConnAckDecoder()); + m_decoderMap.put(AbstractMessage.PUBLISH, new PublishDecoder()); } @Override 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 index 8b06003b..d640fc71 100644 --- 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 @@ -19,6 +19,7 @@ public class MQTTEncoder extends MessageToByteEncoder { MQTTEncoder() { m_encoderMap.put(AbstractMessage.CONNECT, new ConnectEncoder()); m_encoderMap.put(AbstractMessage.CONNACK, new ConnAckEncoder()); + m_encoderMap.put(AbstractMessage.PUBLISH, new PublishEncoder()); } @Override diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java new file mode 100644 index 00000000..876252c6 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java @@ -0,0 +1,61 @@ +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; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +class PublishDecoder extends DemuxDecoder { + + private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class); + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + LOG.debug("decode invoked with buffer " + in); + int startPos = in.readerIndex(); + in.resetReaderIndex(); + + //Common decoding part + PublishMessage message = new PublishMessage(); + if (!decodeCommonHeader(message, in)) { + LOG.info("decode ask for more data after " + in); + in.resetReaderIndex(); + return; + } + int remainingLength = message.getRemainingLength(); + + //Topic name + String topic = Utils.decodeString(in); + if (topic == null) { + in.resetReaderIndex(); + return; + } + message.setTopicName(topic); + + if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || + message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) { + message.setMessageID((int)in.readShort()); + } + int stopPos = in.readerIndex(); + + //read the payload + int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1); + if (in.readableBytes() < payloadSize) { + in.resetReaderIndex(); + return; + } + byte[] b = new byte[payloadSize]; + in.readBytes(b); + message.setPayload(b); + + out.add(message); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java new file mode 100644 index 00000000..171b4dcb --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishEncoder.java @@ -0,0 +1,45 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelHandlerContext; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; + +/** + * + * @author andrea + */ +class PublishEncoder extends DemuxEncoder { + + @Override + protected void encode(ChannelHandlerContext ctx, PublishMessage message, ByteBuf out) { + if (message.getQos() == AbstractMessage.QOSType.RESERVED) { + throw new IllegalArgumentException("Found a message with RESERVED Qos"); + } + if (message.getTopicName() == null || message.getTopicName().isEmpty()) { + throw new IllegalArgumentException("Found a message with empty or null topic name"); + } + + ByteBuf variableHeaderBuff = ctx.alloc().buffer(2); + variableHeaderBuff.writeBytes(Utils.encodeString(message.getTopicName())); + if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || + message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE ) { + if (message.getMessageID() == null) { + throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); + } + variableHeaderBuff.writeShort(message.getMessageID()); + } + variableHeaderBuff.writeBytes(message.getPayload()); + int variableHeaderSize = variableHeaderBuff.readableBytes(); + + byte flags = Utils.encodeFlags(message); + + ByteBuf buff = ctx.alloc().buffer(2 + variableHeaderSize); + buff.writeByte(AbstractMessage.PUBLISH << 4 | flags); + buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); + buff.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 f0814bb3..1252e22c 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 @@ -4,6 +4,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.CorruptedFrameException; import java.io.UnsupportedEncodingException; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.slf4j.LoggerFactory; /** @@ -131,4 +132,28 @@ public class Utils { out.writeBytes(raw); return out; } + + /** + * Return the number of bytes to encode the given remaining length value + */ + static int numBytesToEncode(int len) { + if (0 <= len && len <= 127) return 1; + if (128 <= len && len <= 16383) return 2; + if (16384 <= len && len <= 2097151) return 3; + if (2097152 <= len && len <= 268435455) return 4; + throw new IllegalArgumentException("value shoul be in the range [0..268435455]"); + } + + static byte encodeFlags(AbstractMessage message) { + byte flags = 0; + if (message.isDupFlag()) { + flags |= 0x08; + } + if (message.isRetainFlag()) { + flags |= 0x01; + } + + flags |= ((message.getQos().ordinal() & 0x03) << 1); + return flags; + } } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java index 21235a81..b3c7e40d 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java @@ -32,10 +32,6 @@ public class ConnectEncoderTest { 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 diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java new file mode 100644 index 00000000..c935d3cf --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishDecoderTest.java @@ -0,0 +1,167 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.util.ArrayList; +import java.util.List; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class PublishDecoderTest { + ByteBuf m_buff; + PublishDecoder m_msgdec; + List m_results; + + @Before + public void setUp() { + m_msgdec = new PublishDecoder(); + m_results = new ArrayList(); + } + + @Test + public void testHeader() throws Exception { + m_buff = Unpooled.buffer(14); + initHeader(m_buff); + + //Excercise + m_msgdec.decode(null, m_buff, m_results); + + assertFalse(m_results.isEmpty()); + PublishMessage message = (PublishMessage)m_results.get(0); + assertNotNull(message); + assertEquals("Fake Topic", message.getTopicName()); + assertNull(message.getMessageID()); + assertEquals(AbstractMessage.PUBLISH, message.getMessageType()); + } + + @Test + public void testHeaderWithMessageID() throws Exception { + m_buff = Unpooled.buffer(14); + int messageID = 123; + initHeaderWithMessageID(m_buff, messageID); + + //Exercise + m_msgdec.decode(null, m_buff, m_results); + + assertFalse(m_results.isEmpty()); + PublishMessage message = (PublishMessage)m_results.get(0); + assertNotNull(message); + assertEquals("Fake Topic", message.getTopicName()); + assertEquals(messageID, (int) message.getMessageID()); + } + + + @Test + public void testHeaderWithMessageID_Payload() throws Exception { + m_buff = Unpooled.buffer(14); + int messageID = 123; + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + initHeaderWithMessageID_Payload(m_buff, messageID, payload); + + //Exercise + m_msgdec.decode(null, m_buff, m_results); + + assertFalse(m_results.isEmpty()); + PublishMessage message = (PublishMessage)m_results.get(0); + assertNotNull(message); + assertEquals("Fake Topic", message.getTopicName()); + assertEquals(messageID, (int) message.getMessageID()); + TestUtils.verifyEquals(payload, message.getPayload()); + } + + @Test + public void testBugOnRealCase() throws Exception { + byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght + 0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] string 2 len + 6 content + 0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding + 0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64}; + m_buff = Unpooled.buffer(overallMessage.length); + m_buff.writeBytes(overallMessage); + + //Exercise + m_msgdec.decode(null, m_buff, m_results); + + assertFalse(m_results.isEmpty()); + PublishMessage message = (PublishMessage)m_results.get(0); + assertNotNull(message); + } + + @Test + public void testDecodeBigContent() throws Exception { + int size = 129; + ByteBuf payload = TestUtils.generateRandomPayload(size); + + ByteBuf firstPublish = generatePublishQoS0(payload); + ByteBuf secondPublish = generatePublishQoS0(TestUtils.generateRandomPayload(size)); + + ByteBuf doubleMessageBuf = Unpooled.buffer(size * 2); + doubleMessageBuf.writeBytes(firstPublish).writeBytes(secondPublish); + + + //Exercise + m_msgdec.decode(null, doubleMessageBuf, m_results); + + assertFalse(m_results.isEmpty()); + PublishMessage message = (PublishMessage)m_results.get(0); + assertNotNull(message); + m_results.clear(); + + m_msgdec.decode(null, doubleMessageBuf, m_results); + + assertFalse(m_results.isEmpty()); + message = (PublishMessage)m_results.get(0); + assertNotNull(message); + } + + private void initHeader(ByteBuf buff) throws IllegalAccessException { + ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic")); + buff.clear().writeByte(AbstractMessage.PUBLISH << 4).writeBytes(Utils.encodeRemainingLength(tmp.readableBytes())); + //topic name + buff.writeBytes(tmp); + } + + private void initHeaderWithMessageID(ByteBuf buff, int messageID) throws IllegalAccessException { + ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic")); + tmp.writeShort(messageID); + buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x02) //set Qos to 1 + .writeBytes(Utils.encodeRemainingLength(tmp.readableBytes())); + //topic name + buff.writeBytes(tmp); + } + + private void initHeaderWithMessageID_Payload(ByteBuf buff, int messageID, byte[] payload) throws IllegalAccessException { + ByteBuf tmp = Unpooled.buffer(4).writeBytes(Utils.encodeString("Fake Topic")); + tmp.writeShort(messageID); + tmp.writeBytes(payload); + buff.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x02) //set Qos to 1 + .writeBytes(Utils.encodeRemainingLength(tmp.readableBytes())); + //topic name + buff.writeBytes(tmp); + } + + private ByteBuf generatePublishQoS0(ByteBuf payload) throws IllegalAccessException { + int size = payload.capacity(); + ByteBuf messageBody = Unpooled.buffer(size); + messageBody.writeBytes(Utils.encodeString("/topic")); + + //ONLY for QoS > 1 Utils.writeWord(messageBody, messageID); + messageBody.writeBytes(payload); + + ByteBuf completeMsg = Unpooled.buffer(size); + completeMsg.clear().writeByte(AbstractMessage.PUBLISH << 4 | 0x00) //set Qos to 0 + .writeBytes(Utils.encodeRemainingLength(messageBody.readableBytes())); + completeMsg.writeBytes(messageBody); + + return completeMsg; + } +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java new file mode 100644 index 00000000..a7216af1 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PublishEncoderTest.java @@ -0,0 +1,175 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.parser.netty.TestUtils.*; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; + +/** + * + * @author andrea + */ +public class PublishEncoderTest { + PublishEncoder m_encoder = new PublishEncoder(); + ChannelHandlerContext m_mockedContext; + ByteBuf m_out; + + @Before + public void setUp() { + //mock the ChannelHandlerContext to return an UnpooledAllocator + m_mockedContext = mockChannelHandler(); + m_out = Unpooled.buffer(); + } + + + @Test + public void testEncodeWithQos_0_noMessageID() throws Exception { + String topic = "/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + + //Verify + assertEquals(0x30, m_out.readByte()); //1 byte + //(2+7) topic + 3 payload + assertEquals(12, m_out.readByte()); //remaining length + + //Variable part + verifyString(topic, m_out); + verifyBuff(payload.length, payload, m_out); + } + + + @Test + public void testEncodeWithQos_1_MessageID() throws Exception { + String topic = "/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + + //Verify + assertEquals(0x32, m_out.readByte()); //1 byte + //(2+7) topic + 2 messageID + 3 payload + assertEquals(14, m_out.readByte()); //remaining length + + //Variable part + verifyString(topic, m_out); + assertEquals(0, m_out.readByte()); //MessageID MSB + assertEquals(1, m_out.readByte()); //MessageID LSB + verifyBuff(payload.length, payload, m_out); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncodeWithQos_1_noMessageID_fail() throws Exception { + String topic = "pictures/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setTopicName(topic); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + ByteBuf out = Unpooled.buffer(); + + //Exercise + m_encoder.encode(m_mockedContext, msg, out); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_null_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(null); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_qos_reserved() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.RESERVED); + msg.setMessageID(1); + msg.setTopicName(null); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + ByteBuf out = Unpooled.buffer(); + + //Exercise + m_encoder.encode(m_mockedContext, msg, out); + } + + @Test + public void testBugMissedMessageIDEncoding() throws Exception{ + String topic = "/topic"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + byte[] payload = "Test my payload".getBytes(); + msg.setPayload(payload); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + + //Verify + //2 byte header + (2+6) topic name + [2 message ID if QoQ <> 0]+ payload + int size = m_out.readableBytes(); + assertEquals(10+payload.length, size); + assertEquals(0x30, m_out.readByte()); //1 byte + assertEquals(23, m_out.readByte()); //1 byte the length + } + + @Test + public void testEncodeBigMessage() throws Exception { + byte[] payload = new byte[256]; + + TestUtils.generateRandomPayload(256).readBytes(payload); + + String topic = "/topic"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + msg.setPayload(payload); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + + int size = m_out.readableBytes(); + assertEquals(11+payload.length, size); + assertEquals(0x30, m_out.readByte()); //1 byte + } + +} 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 index 14242230..29a93411 100644 --- 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 @@ -56,4 +56,20 @@ public class TestUtils { when(m_mockedContext.alloc()).thenReturn(allocator); return m_mockedContext; } + + static void verifyEquals(byte[] expected, byte[] found) { + assertEquals(expected.length, found.length); + for (int i = 0; i < expected.length; i++) { + assertEquals(expected[i], found[i]); + } + } + + + public static ByteBuf generateRandomPayload(int size) { + ByteBuf payloadBuffer = Unpooled.buffer(size); + for (int i = 0; i < size; i++) { + payloadBuffer.writeByte((byte) (Math.random() * 255)); + } + return payloadBuffer; + } } 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 adfbc366..e2a257a2 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 @@ -7,6 +7,8 @@ import org.junit.Before; import org.junit.Test; import static org.junit.Assert.*; import static org.dna.mqtt.moquette.parser.netty.TestUtils.*; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; /** * @@ -78,4 +80,14 @@ public class UtilsTest { 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)); } + + @Test + public void testEncodeFlags() { + UnsubscribeMessage msg = new UnsubscribeMessage(); + msg.setRetainFlag(true); + msg.setQos(AbstractMessage.QOSType.MOST_ONE); + + //Exercise + assertEquals(1, Utils.encodeFlags(msg)); + } }