From 52402536d99336dd00dae114fa60bab2233bbebb Mon Sep 17 00:00:00 2001 From: andrea Date: Thu, 12 Sep 2013 16:28:51 +0200 Subject: [PATCH] Added encode/decode for Unsubscribe message --- .../moquette/parser/netty/MQTTDecoder.java | 1 + .../moquette/parser/netty/MQTTEncoder.java | 1 + .../parser/netty/UnsubscribeDecoder.java | 43 ++++++++ .../parser/netty/UnsubscribeEncoder.java | 42 ++++++++ .../parser/netty/UnsubscribeDecoderTest.java | 95 ++++++++++++++++++ .../parser/netty/UnsubscribeEncoderTest.java | 99 +++++++++++++++++++ 6 files changed, 281 insertions(+) create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java create mode 100644 netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java create mode 100644 netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.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 6730b745..fc5659eb 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 @@ -24,6 +24,7 @@ public class MQTTDecoder extends ByteToMessageDecoder { m_decoderMap.put(AbstractMessage.PUBACK, new PubAckDecoder()); m_decoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeDecoder()); m_decoderMap.put(AbstractMessage.SUBACK, new SubAckDecoder()); + m_decoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeDecoder()); } @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 91d1a528..92b8620b 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 @@ -23,6 +23,7 @@ public class MQTTEncoder extends MessageToByteEncoder { m_encoderMap.put(AbstractMessage.PUBACK, new PubAckEncoder()); m_encoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeEncoder()); m_encoderMap.put(AbstractMessage.SUBACK, new SubAckEncoder()); + m_encoderMap.put(AbstractMessage.UNSUBSCRIBE, new UnsubscribeEncoder()); } @Override diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java new file mode 100644 index 00000000..6aba89e2 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoder.java @@ -0,0 +1,43 @@ +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.util.List; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; + +/** + * + * @author andrea + */ +class UnsubscribeDecoder extends DemuxDecoder { + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + //Common decoding part + in.resetReaderIndex(); + UnsubscribeMessage message = new UnsubscribeMessage(); + if (!decodeCommonHeader(message, in)) { + in.resetReaderIndex(); + return; + } + + //check qos level + if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) { + throw new CorruptedFrameException("Found an Usubscribe message with qos other than LEAST_ONE, was: " + message.getQos()); + } + + int start = in.readerIndex(); + //read messageIDs + message.setMessageID(in.readUnsignedShort()); + int readed = in.readerIndex()- start; + while (readed < message.getRemainingLength()) { + message.addTopic(Utils.decodeString(in)); + readed = in.readerIndex()- start; + } + + out.add(message); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java new file mode 100644 index 00000000..3800264a --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoder.java @@ -0,0 +1,42 @@ +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.UnsubscribeMessage; + + +/** + * + * @author andrea + */ +class UnsubscribeEncoder extends DemuxEncoder { + + @Override + protected void encode(ChannelHandlerContext chc, UnsubscribeMessage message, ByteBuf out) { + if (message.topics().isEmpty()) { + throw new IllegalArgumentException("Found an unsubscribe message with empty topics"); + } + + if (message.getQos() != AbstractMessage.QOSType.LEAST_ONE) { + throw new IllegalArgumentException("Expected a message with QOS 1, found " + message.getQos()); + } + + ByteBuf variableHeaderBuff = chc.alloc().buffer(4); + variableHeaderBuff.writeShort(message.getMessageID()); + for (String topic : message.topics()) { + variableHeaderBuff.writeBytes(Utils.encodeString(topic)); + } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + byte flags = Utils.encodeFlags(message); + ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize); + + buff.writeByte(AbstractMessage.UNSUBSCRIBE << 4 | flags); + buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); + buff.writeBytes(variableHeaderBuff); + + out.writeBytes(buff); + } + +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java new file mode 100644 index 00000000..8e465ea5 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeDecoderTest.java @@ -0,0 +1,95 @@ +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.util.ArrayList; +import java.util.List; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; +import static org.junit.Assert.*; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class UnsubscribeDecoderTest { + ByteBuf m_buff; + UnsubscribeDecoder m_msgdec; + List m_results; + + @Before + public void setUp() { + m_msgdec = new UnsubscribeDecoder(); + m_results = new ArrayList(); + } + + + @Test(expected = CorruptedFrameException.class) + public void testBadQos() throws Exception { + m_buff = Unpooled.buffer(2); + initHeaderBadQos(m_buff); + + //Excercise + m_msgdec.decode(null, m_buff, m_results); + } + + @Test + public void testMultiTopic() throws Exception { + m_buff = Unpooled.buffer(4); + String topic1 = "a/b"; + String topic2 = "c/d/e"; + initMultiTopic(m_buff, 123, topic1, topic2); + + //Excercise + m_msgdec.decode(null, m_buff, m_results); + + //Verify + assertFalse(m_results.isEmpty()); + UnsubscribeMessage message = (UnsubscribeMessage)m_results.get(0); + assertNotNull(message); + assertEquals(2, message.topics().size()); + assertEquals(topic1, message.topics().get(0)); + assertEquals(topic2, message.topics().get(1)); + assertEquals(AbstractMessage.UNSUBSCRIBE, message.getMessageType()); + } + + + @Test + public void testDecodeSingleTopic_bug() throws Exception { + //A2 0C 00 01 00 06 2F 74 6F 70 69 63 //12 byte + byte[] overallMessage = new byte[] {(byte)0xA2, 0x0A, //fixed header + 0x00, 0x01, //MSG ID + 0x00, 0x06, 0x2F, 0x74, 0x6F, 0x70, 0x69, 0x63}; //"/topic" string + m_buff = Unpooled.buffer(overallMessage.length); + m_buff.writeBytes(overallMessage); + + m_msgdec.decode(null, m_buff, m_results); + + assertFalse(m_results.isEmpty()); + UnsubscribeMessage message = (UnsubscribeMessage)m_results.get(0); + assertNotNull(message); + assertEquals(0x01, (int)message.getMessageID()); + } + + + private void initHeaderBadQos(ByteBuf buff) { + buff.clear().writeByte(AbstractMessage.UNSUBSCRIBE << 4).writeByte(0); + } + + private void initMultiTopic(ByteBuf buff, int messageID, String... topics) throws IllegalAccessException { + ByteBuf topicBuffer = Unpooled.buffer(4); + topicBuffer.writeShort(messageID); + for (String topic : topics) { + topicBuffer.writeBytes(Utils.encodeString(topic)); + } + + buff.clear().writeByte(AbstractMessage.UNSUBSCRIBE << 4 | (byte)0x02) + .writeBytes(Utils.encodeRemainingLength(topicBuffer.readableBytes())); + buff.writeBytes(topicBuffer); + } + +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java new file mode 100644 index 00000000..7e900549 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UnsubscribeEncoderTest.java @@ -0,0 +1,99 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import static org.dna.mqtt.moquette.parser.netty.TestUtils.mockChannelHandler; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; +import static org.junit.Assert.assertEquals; +import org.junit.Before; +import org.junit.Test; +import static org.dna.mqtt.moquette.parser.netty.TestUtils.*; + +/** + * + * @author andrea + */ +public class UnsubscribeEncoderTest { + UnsubscribeEncoder m_encoder = new UnsubscribeEncoder(); + ChannelHandlerContext m_mockedContext; + ByteBuf m_out; + UnsubscribeMessage m_msg; + + @Before + public void setUp() { + //mock the ChannelHandlerContext to return an UnpooledAllocator + m_mockedContext = mockChannelHandler(); + m_out = Unpooled.buffer(); + m_msg = new UnsubscribeMessage(); + m_msg.setMessageID(0xAABB); + } + + @Test + public void testEncodeWithSingleTopic() throws Exception { + m_msg.setQos(AbstractMessage.QOSType.LEAST_ONE); + + //variable part + String topic1 = "/topic"; + m_msg.addTopic(topic1); + + //Exercise + m_encoder.encode(m_mockedContext, m_msg, m_out); + + //Verify + assertEquals((byte)0xA2, (byte)m_out.readByte()); //1 byte + //2 messageID + 2 length + 6 chars = 10 + assertEquals(10, m_out.readByte()); //remaining length + + //verify M1ssageID + assertEquals((byte)0xAA, m_out.readByte()); + assertEquals((byte)0xBB, m_out.readByte()); + + //Variable part + verifyString(topic1, m_out); + } + + + @Test + public void testEncodeWithMultiTopic() throws Exception { + m_msg.setQos(AbstractMessage.QOSType.LEAST_ONE); + + //variable part + String topic1 = "a/b"; + String topic2 = "a/b/c"; + m_msg.addTopic(topic1); + m_msg.addTopic(topic2); + + //Exercise + m_encoder.encode(m_mockedContext, m_msg, m_out); + + //Verify + assertEquals((byte)0xA2, (byte)m_out.readByte()); //1 byte + assertEquals(14, m_out.readByte()); //remaining length + + //verify M1ssageID + assertEquals((byte)0xAA, m_out.readByte()); + assertEquals((byte)0xBB, m_out.readByte()); + + //Variable part + verifyString(topic1, m_out); + verifyString(topic2, m_out); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_topics() throws Exception { + m_msg.setQos(AbstractMessage.QOSType.LEAST_ONE); + + //Exercise + m_encoder.encode(m_mockedContext, m_msg, m_out); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_badQos() throws Exception { + m_msg.setQos(AbstractMessage.QOSType.EXACTLY_ONCE); + + //Exercise + m_encoder.encode(m_mockedContext, m_msg, m_out); + } +}