Added encode/decode for PubAck message

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java

@@ -70,7 +70,7 @@ public class ConnectDecoder extends DemuxDecoder {
 
         //Keep Alive timer 2 bytes
         //int keepAlive = Utils.readWord(in);
-        int keepAlive = in.readShort();
+        int keepAlive = in.readUnsignedShort();
         message.setKeepAlive(keepAlive);
 
         if (remainingLength == 12) {

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java

@@ -21,6 +21,7 @@ public class MQTTDecoder extends ByteToMessageDecoder {
        m_decoderMap.put(AbstractMessage.CONNECT, new ConnectDecoder());
        m_decoderMap.put(AbstractMessage.CONNACK, new ConnAckDecoder());
        m_decoderMap.put(AbstractMessage.PUBLISH, new PublishDecoder());
+       m_decoderMap.put(AbstractMessage.PUBACK, new PubAckDecoder());
     }
 
     @Override

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java

@@ -20,6 +20,7 @@ public class MQTTEncoder extends MessageToByteEncoder<AbstractMessage> {
        m_encoderMap.put(AbstractMessage.CONNECT, new ConnectEncoder());
        m_encoderMap.put(AbstractMessage.CONNACK, new ConnAckEncoder());
        m_encoderMap.put(AbstractMessage.PUBLISH, new PublishEncoder());
+       m_encoderMap.put(AbstractMessage.PUBACK, new PubAckEncoder());
     }
     
     @Override

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MessageIDDecoder.java

@@ -0,0 +1,31 @@
+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.MessageIDMessage;
+
+/**
+ *
+ * @author andrea
+ */
+abstract class MessageIDDecoder extends DemuxDecoder {
+    
+    protected abstract MessageIDMessage createMessage();
+
+    @Override
+    void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        in.resetReaderIndex();
+        //Common decoding part
+        MessageIDMessage message = createMessage();
+        if (!decodeCommonHeader(message, in)) {
+            in.resetReaderIndex();
+            return;
+        }
+        
+        //read  messageIDs
+        message.setMessageID(in.readUnsignedShort());
+        out.add(message);
+    }
+    
+}

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckDecoder.java

@@ -0,0 +1,17 @@
+package org.dna.mqtt.moquette.parser.netty;
+
+import org.dna.mqtt.moquette.proto.messages.MessageIDMessage;
+import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
+
+/**
+ *
+ * @author andrea
+ */
+class PubAckDecoder extends MessageIDDecoder {
+
+    @Override
+    protected MessageIDMessage createMessage() {
+        return new PubAckMessage();
+    }
+    
+}

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java

@@ -0,0 +1,23 @@
+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.PubAckMessage;
+
+/**
+ *
+ * @author andrea
+ */
+class PubAckEncoder extends DemuxEncoder<PubAckMessage> {
+
+    @Override
+    protected void encode(ChannelHandlerContext chc, PubAckMessage msg, ByteBuf out) {
+        ByteBuf buff = chc.alloc().buffer(4);
+        buff.writeByte(AbstractMessage.PUBACK << 4);
+        buff.writeBytes(Utils.encodeRemainingLength(2));
+        buff.writeShort(msg.getMessageID());
+        out.writeBytes(buff);
+    }
+    
+}

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java

@@ -41,7 +41,7 @@ class PublishDecoder extends DemuxDecoder {
         
         if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || 
                 message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) {
-            message.setMessageID((int)in.readShort());
+            message.setMessageID(in.readUnsignedShort());
         }
         int stopPos = in.readerIndex();
         

netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java

@@ -101,7 +101,7 @@ public class Utils {
             return null;
         }
         //int strLen = Utils.readWord(in);
-        int strLen = in.readShort();
+        int strLen = in.readUnsignedShort();
         if (in.readableBytes() < strLen) {
             return null;
         }

netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PubAckDecoderTest.java

@@ -0,0 +1,51 @@
+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.PubAckMessage;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ *
+ * @author andrea
+ */
+public class PubAckDecoderTest {
+    ByteBuf m_buff;
+    PubAckDecoder m_msgdec;
+    
+    @Before
+    public void setUp() {
+        m_msgdec = new PubAckDecoder();
+    }
+    
+    @Test
+    public void testHeader() throws Exception {
+        m_buff = Unpooled.buffer(14);
+        int messageId = 0xAABB;
+        initHeader(m_buff, messageId);
+        List<Object> results = new ArrayList<Object >();
+        
+        //Excercise
+        m_msgdec.decode(null, m_buff, results);
+        
+        assertFalse(results.isEmpty());
+        PubAckMessage message = (PubAckMessage)results.get(0); 
+        assertNotNull(message);
+        assertEquals(messageId, message.getMessageID().intValue());
+        assertEquals(AbstractMessage.PUBACK, message.getMessageType());
+    }
+    
+    private void initHeader(ByteBuf buff, int messageID) {
+        buff.clear().writeByte(AbstractMessage.PUBACK << 4).writeByte(2);
+        
+        //return code
+        buff.writeShort(messageID);
+    }
+}

netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoderTest.java

@@ -0,0 +1,42 @@
+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.Test;
+import org.junit.Before;
+import static org.junit.Assert.*;
+import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
+import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
+
+/**
+ *
+ * @author andrea
+ */
+public class PubAckEncoderTest {
+    ChannelHandlerContext m_mockedContext;
+         
+    @Before
+    public void setUp() {
+        //mock the ChannelHandlerContext to return an UnpooledAllocator
+        m_mockedContext = mockChannelHandler();
+    }
+    
+    @Test
+    public void testHeaderEncode() throws Exception {
+        int messageID = 0xAABB;
+        PubAckEncoder encoder = new PubAckEncoder();
+        PubAckMessage msg = new PubAckMessage();
+        msg.setMessageID(messageID);
+        ByteBuf out = Unpooled.buffer();
+        
+        //Exercise
+        encoder.encode(m_mockedContext, msg, out);
+        
+        //Verify
+        assertEquals(0x40, out.readByte()); //1 byte
+        assertEquals(0x02, out.readByte()); //2 byte, length
+        assertEquals((byte)0xAA, out.readByte());
+        assertEquals((byte)0xBB, out.readByte());
+    }
+}