Added encode/decode for Unsubscribe message

2013-09-12 16:28:51 +02:00
parent daa637c557
commit 52402536d9
6 changed files with 281 additions and 0 deletions
@@ -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
@@ -23,6 +23,7 @@ public class MQTTEncoder extends MessageToByteEncoder<AbstractMessage> {
       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
@@ -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<Object> 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);
+    }
+    
+}
@@ -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<UnsubscribeMessage> {
+
+    @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);
+    }
+    
+}
@@ -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<Object> m_results;
+    
+    @Before
+    public void setUp() {
+        m_msgdec = new UnsubscribeDecoder();
+        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);
+        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);
+    }
+    
+}
@@ -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);
+    }
+}