diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java index a21b1f53..3ecbcf25 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java +++ b/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) { diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java index 8cfe6e9d..469c0bac 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java +++ b/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 diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java index d640fc71..d34892f2 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java @@ -20,6 +20,7 @@ public class MQTTEncoder extends MessageToByteEncoder { 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 diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MessageIDDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MessageIDDecoder.java new file mode 100644 index 00000000..41a71164 --- /dev/null +++ b/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 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); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckDecoder.java new file mode 100644 index 00000000..7aa20ee2 --- /dev/null +++ b/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(); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoder.java new file mode 100644 index 00000000..56d608a4 --- /dev/null +++ b/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 { + + @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); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java index 876252c6..9a29ff14 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java +++ b/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(); diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java index 1252e22c..68bb0a53 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java +++ b/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; } diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PubAckDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PubAckDecoderTest.java new file mode 100644 index 00000000..0cea6581 --- /dev/null +++ b/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 results = new ArrayList(); + + //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); + } +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/PubAckEncoderTest.java new file mode 100644 index 00000000..9b83c285 --- /dev/null +++ b/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()); + } +}