Fixed issue 4, bad calculation of remaining length in big PublishDecoder.

parser/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java

@@ -7,6 +7,8 @@ import org.apache.mina.filter.codec.demux.MessageDecoderResult;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 import org.dna.mqtt.moquette.proto.messages.PublishMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  *
@@ -14,16 +16,20 @@ import org.dna.mqtt.moquette.proto.messages.PublishMessage;
  */
 public class PublishDecoder extends MqttDecoder {
     
+    private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class);
+    
     public MessageDecoderResult decodable(IoSession session, IoBuffer in) {
         return Utils.checkDecodable(AbstractMessage.PUBLISH, in);
     }
 
     public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
+        LOG.debug("decode invoked with buffer " + in);
         int startPos = in.position();
 
         //Common decoding part
         PublishMessage message = new PublishMessage();
         if (decodeCommonHeader(message, in) == NEED_DATA) {
+            LOG.info("decode ask for more data after " + in);
             return NEED_DATA;
         }
         int remainingLength = message.getRemainingLength();
@@ -42,11 +48,11 @@ public class PublishDecoder extends MqttDecoder {
         int stopPos = in.position();
         
         //read the payload
-        int payloadSize = remainingLength - (stopPos - startPos - 2);
+        int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1);
-        byte[] b = new byte[payloadSize];
         if (in.remaining() < payloadSize) {
             return NEED_DATA;
         }
+        byte[] b = new byte[payloadSize];
         in.get(b);
         message.setPayload(b);
         

parser/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java

@@ -37,7 +37,7 @@ public class PublishEncoder implements MessageEncoder<PublishMessage> {
         
         byte flags = Utils.encodeFlags(message);
         
-        IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize);
+        IoBuffer buff = IoBuffer.allocate(2 + variableHeaderSize).setAutoExpand(true);;
         buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags));
         buff.put(Utils.encodeRemainingLength(variableHeaderSize));
         buff.put(variableHeaderBuff).flip();

parser/src/main/java/org/dna/mqtt/moquette/proto/Utils.java

@@ -54,6 +54,17 @@ public class Utils {
         return value;
     }
     
+    /**
+     * Return the number of bytes to encode the gicen remaining length value
+     */
+    static int numBytesToEncode(int len) {
+        if (0 <= len && len <= 127) return 1;
+        if (128 <= len && len <= 16383) return 2;
+        if (16384 <= len && len <= 2097151) return 3;
+        if (2097152 <= len && len <= 268435455) return 4;
+        throw new IllegalArgumentException("value shoul be in the range [0..268435455]");
+    }
+    
     /**
      * Encode the value in the format defined in specification as variable length
      * array.

parser/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java

@@ -1,5 +1,9 @@
 package org.dna.mqtt.moquette.proto;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
 import org.apache.mina.core.buffer.IoBuffer;
 import org.apache.mina.filter.codec.demux.MessageDecoder;
 import org.apache.mina.filter.codec.demux.MessageDecoderResult;
@@ -8,6 +12,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.dna.mqtt.moquette.proto.messages.PublishMessage;
 import static org.junit.Assert.*;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 
 /**
@@ -30,7 +35,7 @@ public class PublishDecoderTest {
     @Test
     public void testDecodable_OK() {
         m_buff.put((byte) (AbstractMessage.PUBLISH << 4))
-                .put((byte)0) //0 length
+                .put((byte) 0) //0 length
                 .flip();
         assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff));
     }
@@ -85,22 +90,49 @@ public class PublishDecoderTest {
         TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().getPayload());
     }
 
-     @Test
+    @Test
-     public void testBugOnRealCase() throws Exception {
+    public void testBugOnRealCase() throws Exception {
-         byte[] overallMessage = new byte[] {0x30, 0x17, //fixed header, 25 byte lenght
+        byte[] overallMessage = new byte[]{0x30, 0x17, //fixed header, 25 byte lenght
-             0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] string 2 len + 6 content
+            0x00, 0x06, 0x2f, 0x74, 0x6f, 0x70, 0x69, 0x63, //[/topic] string 2 len + 6 content
-             0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding
+            0x54, 0x65, 0x73, 0x74, 0x20, 0x6d, 0x79, // [Test my payload] encoding
-             0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
+            0x20, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64};
-         m_buff = IoBuffer.allocate(overallMessage.length).setAutoExpand(true);
+        m_buff = IoBuffer.allocate(overallMessage.length).setAutoExpand(true);
-         m_buff.put(overallMessage);
+        m_buff.put(overallMessage);
-         m_buff.flip();
+        m_buff.flip();
 
-         //Exercise
+        //Exercise
         MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder);
 
         assertNotNull(m_mockProtoDecoder.getMessage());
         assertEquals(MessageDecoder.OK, res);
-     }
+    }
+
+    @Test
+    public void testDecodeBigContent() throws Exception {
+        int size = 129;
+        IoBuffer payload = TestUtils.generateRandomPayload(size);
+
+        IoBuffer firstPublish = generatePublishQoS0(payload);
+        IoBuffer secondPublish = generatePublishQoS0(TestUtils.generateRandomPayload(size));
+
+        IoBuffer doubleMessageBuf = IoBuffer.allocate(size * 2).setAutoExpand(true);
+        doubleMessageBuf.put(firstPublish).put(secondPublish).flip();
+
+
+        //Exercise
+        MessageDecoderResult res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder);
+
+        assertEquals(MessageDecoder.OK, res);
+        PublishMessage pubMsg = m_mockProtoDecoder.getMessage();
+        assertNotNull(pubMsg);
+
+        res = m_msgdec.decode(null, doubleMessageBuf, m_mockProtoDecoder);
+
+        assertNotNull(m_mockProtoDecoder.getMessage());
+        assertEquals(MessageDecoder.OK, res);
+
+        m_buff.flip();
+    }
 
     private void initHeader(IoBuffer buff) throws IllegalAccessException {
         IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip();
@@ -129,4 +161,20 @@ public class PublishDecoderTest {
         //topic name
         buff.put(tmp);
     }
+
+    private IoBuffer generatePublishQoS0(IoBuffer payload) throws IllegalAccessException {
+        int size = payload.capacity();
+        IoBuffer messageBody = IoBuffer.allocate(size).setAutoExpand(true);
+        messageBody.put(Utils.encodeString("/topic"));
+
+        //ONLY for QoS > 1 Utils.writeWord(messageBody, messageID);
+        messageBody.put(payload).flip();
+
+        IoBuffer completeMsg = IoBuffer.allocate(size).setAutoExpand(true);
+        completeMsg.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x00)) //set Qos to 0
+                .put(Utils.encodeRemainingLength(messageBody.remaining()));
+        completeMsg.put(messageBody).flip();
+
+        return completeMsg;
+    }
 }

parser/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java

@@ -1,6 +1,6 @@
 package org.dna.mqtt.moquette.proto;
 
-import org.dna.mqtt.moquette.proto.PublishEncoder;
+import org.apache.mina.core.buffer.IoBuffer;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
 import org.dna.mqtt.moquette.proto.messages.PublishMessage;
 import org.junit.Before;
@@ -142,4 +142,26 @@ public class PublishEncoderTest {
         assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte
         assertEquals(23, m_mockProtoEncoder.getBuffer().get()); //1 byte the length
     }
+    
+    @Test
+    public void testEncodeBigMessage() throws Exception {
+        byte[] payload = new byte[256];
+        
+        TestUtils.generateRandomPayload(256).get(payload);
+        
+        String topic = "/topic";
+        PublishMessage msg = new PublishMessage();
+        msg.setQos(QOSType.MOST_ONE);
+        msg.setTopicName(topic);
+
+        //variable part
+        msg.setPayload(payload);
+
+        //Exercise
+        m_encoder.encode(null, msg, m_mockProtoEncoder);
+        
+        int size = m_mockProtoEncoder.getBuffer().remaining();
+        assertEquals(11+payload.length, size);
+        assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte
+    }
 }

parser/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java

@@ -1,5 +1,9 @@
 package org.dna.mqtt.moquette.proto;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import org.apache.mina.core.buffer.IoBuffer;
 import org.apache.mina.core.filterchain.IoFilter.NextFilter;
@@ -94,4 +98,27 @@ public class TestUtils {
             assertEquals(expected[i], found[i]);
         }
     }
+    
+    
+    public static IoBuffer generateRandomPayload(int size) {
+        IoBuffer payloadBuffer = IoBuffer.allocate(size);
+        for (int i = 0; i < size; i++) {
+            payloadBuffer.put((byte) (Math.random() * 255));
+        }
+        payloadBuffer.flip();
+        return payloadBuffer;
+    }
+    
+    public static IoBuffer loadFile(String path) throws FileNotFoundException, IOException {
+        File f = new File(path);
+        long size = f.length();
+        byte[] payload = new byte[(int) size];
+
+        FileInputStream fr = new FileInputStream(f);
+        fr.read(payload);
+
+        IoBuffer payloadBuffer = IoBuffer.allocate((int) size);
+        payloadBuffer.put(payload).flip();
+        return payloadBuffer;
+    }
 }