Added encode/decode for Subscribe message

2013-09-12 15:16:55 +02:00
parent 1b1cc715be
commit c3f5a69d68
6 changed files with 246 additions and 0 deletions
@@ -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<Object> m_results;
+    
+    @Before
+    public void setUp() {
+        m_msgdec = new SubscribeDecoder();
+        m_results = new ArrayList<Object >();
+    }
+    
+    @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);
+    }
+}
@@ -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);
+    }
+}