From c3f5a69d68e3e148971687021b747ec82b5444d6 Mon Sep 17 00:00:00 2001 From: andrea Date: Thu, 12 Sep 2013 15:16:55 +0200 Subject: [PATCH] Added encode/decode for Subscribe message --- .../moquette/parser/netty/MQTTDecoder.java | 1 + .../moquette/parser/netty/MQTTEncoder.java | 1 + .../parser/netty/SubscribeDecoder.java | 53 +++++++++++++ .../parser/netty/SubscribeEncoder.java | 42 ++++++++++ .../parser/netty/SubscribeDecoderTest.java | 71 +++++++++++++++++ .../parser/netty/SubscribeEncoderTest.java | 78 +++++++++++++++++++ 6 files changed, 246 insertions(+) create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoder.java create mode 100644 netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java create mode 100644 netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java create mode 100644 netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.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 469c0bac..f7d2792f 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 @@ -22,6 +22,7 @@ public class MQTTDecoder extends ByteToMessageDecoder { m_decoderMap.put(AbstractMessage.CONNACK, new ConnAckDecoder()); m_decoderMap.put(AbstractMessage.PUBLISH, new PublishDecoder()); m_decoderMap.put(AbstractMessage.PUBACK, new PubAckDecoder()); + m_decoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeDecoder()); } @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 d34892f2..3153f208 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 @@ -21,6 +21,7 @@ public class MQTTEncoder extends MessageToByteEncoder { m_encoderMap.put(AbstractMessage.CONNACK, new ConnAckEncoder()); m_encoderMap.put(AbstractMessage.PUBLISH, new PublishEncoder()); m_encoderMap.put(AbstractMessage.PUBACK, new PubAckEncoder()); + m_encoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeEncoder()); } @Override diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoder.java new file mode 100644 index 00000000..efdb84e4 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoder.java @@ -0,0 +1,53 @@ +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.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; + +/** + * + * @author andrea + */ +class SubscribeDecoder extends DemuxDecoder { + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + //Common decoding part + SubscribeMessage message = new SubscribeMessage(); + in.resetReaderIndex(); + if (!decodeCommonHeader(message, in)) { + in.resetReaderIndex(); + return; + } + + //check qos level + if (message.getQos() != QOSType.LEAST_ONE) { + throw new CorruptedFrameException("Received Subscribe 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()) { + decodeSubscription(in, message); + readed = in.readerIndex()- start; + } + + out.add(message); + } + + /** + * Populate the message with couple of Qos, topic + */ + private void decodeSubscription(ByteBuf in, SubscribeMessage message) throws UnsupportedEncodingException { + String topic = Utils.decodeString(in); + byte qos = (byte)(in.readByte() & 0x03); + message.addSubscription(new SubscribeMessage.Couple(qos, topic)); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.java new file mode 100644 index 00000000..c97749f7 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoder.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.SubscribeMessage; + +/** + * + * @author andrea + */ +class SubscribeEncoder extends DemuxEncoder { + + @Override + protected void encode(ChannelHandlerContext chc, SubscribeMessage message, ByteBuf out) { + if (message.subscriptions().isEmpty()) { + throw new IllegalArgumentException("Found a subscribe 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 (SubscribeMessage.Couple c : message.subscriptions()) { + variableHeaderBuff.writeBytes(Utils.encodeString(c.getTopic())); + variableHeaderBuff.writeByte(c.getQos()); + } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + byte flags = Utils.encodeFlags(message); + ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize); + + buff.writeByte(AbstractMessage.SUBSCRIBE << 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/SubscribeDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java new file mode 100644 index 00000000..f140adf3 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeDecoderTest.java @@ -0,0 +1,71 @@ +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 static org.junit.Assert.*; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class SubscribeDecoderTest { + ByteBuf m_buff; + SubscribeDecoder m_msgdec; + List m_results; + + @Before + public void setUp() { + m_msgdec = new SubscribeDecoder(); + 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); + SubscribeMessage.Couple c1 = new SubscribeMessage.Couple((byte)2, "a/b"); + SubscribeMessage.Couple c2 = new SubscribeMessage.Couple((byte)1, "c/d/e"); + initMultiTopic(m_buff, 123, c1, c2); + + //Excercise + m_msgdec.decode(null, m_buff, m_results); + + //Verify + assertFalse(m_results.isEmpty()); + SubscribeMessage message = (SubscribeMessage)m_results.get(0); + assertEquals(2, message.subscriptions().size()); + assertEquals(AbstractMessage.SUBSCRIBE, message.getMessageType()); + } + + private void initHeaderBadQos(ByteBuf buff) { + buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 4).writeByte(0); + } + + private void initMultiTopic(ByteBuf buff, int messageID, SubscribeMessage.Couple... topics) throws IllegalAccessException { + ByteBuf topicBuffer = Unpooled.buffer(4); + topicBuffer.writeShort(messageID); + for (SubscribeMessage.Couple couple : topics) { + topicBuffer.writeBytes(Utils.encodeString(couple.getTopic())); + topicBuffer.writeByte(couple.getQos()); + } + + buff.clear().writeByte(AbstractMessage.SUBSCRIBE << 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/SubscribeEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.java new file mode 100644 index 00000000..014abe23 --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubscribeEncoderTest.java @@ -0,0 +1,78 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +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 SubscribeEncoderTest { + SubscribeEncoder m_encoder = new SubscribeEncoder(); + 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 testEncodeWithMultiTopic() throws Exception { + SubscribeMessage msg = new SubscribeMessage(); + msg.setQos(AbstractMessage.QOSType.LEAST_ONE); + msg.setMessageID(0xAABB); + + //variable part + SubscribeMessage.Couple c1 = new SubscribeMessage.Couple((byte)1, "a/b"); + SubscribeMessage.Couple c2 = new SubscribeMessage.Couple((byte)0, "a/b/c"); + msg.addSubscription(c1); + msg.addSubscription(c2); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + + //Verify + assertEquals((byte)0x82, (byte)m_out.readByte()); //1 byte + assertEquals(16, m_out.readByte()); //remaining length + + //verify MessageID + assertEquals((byte)0xAA, m_out.readByte()); + assertEquals((byte)0xBB, m_out.readByte()); + + //Variable part + verifyString(c1.getTopic(), m_out); + assertEquals(c1.getQos(), m_out.readByte()); + verifyString(c2.getTopic(), m_out); + assertEquals(c2.getQos(), m_out.readByte()); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_subscription() throws Exception { + SubscribeMessage msg = new SubscribeMessage(); + msg.setQos(AbstractMessage.QOSType.LEAST_ONE); + msg.setMessageID(0xAABB); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_badQos() throws Exception { + SubscribeMessage msg = new SubscribeMessage(); + msg.setQos(AbstractMessage.QOSType.EXACTLY_ONCE); + msg.setMessageID(0xAABB); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + } +}