Added encode/decode for Publish message

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

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

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

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

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

@@ -0,0 +1,61 @@
+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.AbstractMessage;
+import org.dna.mqtt.moquette.proto.messages.PublishMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author andrea
+ */
+class PublishDecoder extends DemuxDecoder {
+    
+    private static Logger LOG = LoggerFactory.getLogger(PublishDecoder.class);
+
+    @Override
+    void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        LOG.debug("decode invoked with buffer " + in);
+        int startPos = in.readerIndex();
+        in.resetReaderIndex();
+
+        //Common decoding part
+        PublishMessage message = new PublishMessage();
+        if (!decodeCommonHeader(message, in)) {
+            LOG.info("decode ask for more data after " + in);
+            in.resetReaderIndex();
+            return;
+        }
+        int remainingLength = message.getRemainingLength();
+        
+        //Topic name
+        String topic = Utils.decodeString(in);
+        if (topic == null) {
+            in.resetReaderIndex();
+            return;
+        }
+        message.setTopicName(topic);
+        
+        if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || 
+                message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE) {
+            message.setMessageID((int)in.readShort());
+        }
+        int stopPos = in.readerIndex();
+        
+        //read the payload
+        int payloadSize = remainingLength - (stopPos - startPos - 2) + (Utils.numBytesToEncode(remainingLength) - 1);
+        if (in.readableBytes() < payloadSize) {
+            in.resetReaderIndex();
+            return;
+        }
+        byte[] b = new byte[payloadSize];
+        in.readBytes(b);
+        message.setPayload(b);
+        
+        out.add(message);
+    }
+    
+}

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

@@ -0,0 +1,45 @@
+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.PublishMessage;
+
+/**
+ *
+ * @author andrea
+ */
+class PublishEncoder extends DemuxEncoder<PublishMessage> {
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, PublishMessage message, ByteBuf out) {
+        if (message.getQos() == AbstractMessage.QOSType.RESERVED) {
+            throw new IllegalArgumentException("Found a message with RESERVED Qos");
+        }
+        if (message.getTopicName() == null || message.getTopicName().isEmpty()) {
+            throw new IllegalArgumentException("Found a message with empty or null topic name");
+        }
+        
+        ByteBuf variableHeaderBuff = ctx.alloc().buffer(2);
+        variableHeaderBuff.writeBytes(Utils.encodeString(message.getTopicName()));
+        if (message.getQos() == AbstractMessage.QOSType.LEAST_ONE || 
+            message.getQos() == AbstractMessage.QOSType.EXACTLY_ONCE ) {
+            if (message.getMessageID() == null) {
+                throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted");
+            }
+            variableHeaderBuff.writeShort(message.getMessageID());
+        }
+        variableHeaderBuff.writeBytes(message.getPayload());
+        int variableHeaderSize = variableHeaderBuff.readableBytes();
+        
+        byte flags = Utils.encodeFlags(message);
+        
+        ByteBuf buff = ctx.alloc().buffer(2 + variableHeaderSize);
+        buff.writeByte(AbstractMessage.PUBLISH << 4 | flags);
+        buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize));
+        buff.writeBytes(variableHeaderBuff);
+
+        out.writeBytes(buff);
+    }
+    
+}

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

@@ -4,6 +4,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.CorruptedFrameException;
 import java.io.UnsupportedEncodingException;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import org.slf4j.LoggerFactory;
 
 /**
@@ -131,4 +132,28 @@ public class Utils {
         out.writeBytes(raw);
         return out;
     }
+    
+    /**
+     * Return the number of bytes to encode the given 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]");
+    }
+    
+    static byte encodeFlags(AbstractMessage message) {
+        byte flags = 0;
+        if (message.isDupFlag()) {
+            flags |= 0x08;
+        }
+        if (message.isRetainFlag()) {
+            flags |= 0x01;
+        }
+        
+        flags |= ((message.getQos().ordinal() & 0x03) << 1);
+        return flags;
+    }
 }