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 f7d2792f..6730b745 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 @@ -23,6 +23,7 @@ public class MQTTDecoder extends ByteToMessageDecoder { m_decoderMap.put(AbstractMessage.PUBLISH, new PublishDecoder()); m_decoderMap.put(AbstractMessage.PUBACK, new PubAckDecoder()); m_decoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeDecoder()); + m_decoderMap.put(AbstractMessage.SUBACK, new SubAckDecoder()); } @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 3153f208..91d1a528 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 @@ -22,6 +22,7 @@ public class MQTTEncoder extends MessageToByteEncoder { m_encoderMap.put(AbstractMessage.PUBLISH, new PublishEncoder()); m_encoderMap.put(AbstractMessage.PUBACK, new PubAckEncoder()); m_encoderMap.put(AbstractMessage.SUBSCRIBE, new SubscribeEncoder()); + m_encoderMap.put(AbstractMessage.SUBACK, new SubAckEncoder()); } @Override diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckDecoder.java new file mode 100644 index 00000000..50538e83 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckDecoder.java @@ -0,0 +1,43 @@ +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.SubAckMessage; + +/** + * + * @author andrea + */ +class SubAckDecoder extends DemuxDecoder { + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + //Common decoding part + in.resetReaderIndex(); + SubAckMessage message = new SubAckMessage(); + if (!decodeCommonHeader(message, in)) { + in.resetReaderIndex(); + return; + } + int remainingLength = message.getRemainingLength(); + + //MessageID + message.setMessageID(in.readUnsignedShort()); + remainingLength -= 2; + + //Qos array + if (in.readableBytes() < remainingLength ) { + in.resetReaderIndex(); + return; + } + for (int i = 0; i < remainingLength; i++) { + byte qos = in.readByte(); + message.addType(AbstractMessage.QOSType.values()[qos]); + } + + out.add(message); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java new file mode 100644 index 00000000..6cb0dd48 --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoder.java @@ -0,0 +1,36 @@ +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.SubAckMessage; + +/** + * + * @author andrea + */ +class SubAckEncoder extends DemuxEncoder { + + @Override + protected void encode(ChannelHandlerContext chc, SubAckMessage message, ByteBuf out) { + if (message.types().isEmpty()) { + throw new IllegalArgumentException("Found a suback message with empty topics"); + } + + ByteBuf variableHeaderBuff = chc.alloc().buffer(4); + variableHeaderBuff.writeShort(message.getMessageID()); + for (AbstractMessage.QOSType c : message.types()) { + variableHeaderBuff.writeByte(c.ordinal()); + } + + int variableHeaderSize = variableHeaderBuff.readableBytes(); + ByteBuf buff = chc.alloc().buffer(2 + variableHeaderSize); + + buff.writeByte(AbstractMessage.SUBACK << 4 ); + buff.writeBytes(Utils.encodeRemainingLength(variableHeaderSize)); + buff.writeBytes(variableHeaderBuff); + + out.writeBytes(buff); + } + +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubAckDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubAckDecoderTest.java new file mode 100644 index 00000000..728344bc --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubAckDecoderTest.java @@ -0,0 +1,78 @@ +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.SubAckMessage; +import static org.junit.Assert.*; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class SubAckDecoderTest { + ByteBuf m_buff; + SubAckDecoder m_msgdec; + List m_results; + + @Before + public void setUp() { + m_msgdec = new SubAckDecoder(); + m_results = new ArrayList(); + m_buff = Unpooled.buffer(7); + } + + @Test + public void testBadQos() throws Exception { + initHeaderQos(m_buff, 0xAABB, AbstractMessage.QOSType.LEAST_ONE, AbstractMessage.QOSType.MOST_ONE, AbstractMessage.QOSType.MOST_ONE); + + //Excercise + m_msgdec.decode(null, m_buff, m_results); + + //Verify + assertFalse(m_results.isEmpty()); + SubAckMessage message = (SubAckMessage)m_results.get(0); + assertNotNull(message); + assertEquals(0xAABB, message.getMessageID().intValue()); + List qoses = message.types(); + assertEquals(3, qoses.size()); + assertEquals(AbstractMessage.QOSType.LEAST_ONE, qoses.get(0)); + assertEquals(AbstractMessage.QOSType.MOST_ONE, qoses.get(1)); + assertEquals(AbstractMessage.QOSType.MOST_ONE, qoses.get(2)); + assertEquals(AbstractMessage.SUBACK, message.getMessageType()); + } + + + @Test + public void testBugBadRemainingCalculation() throws Exception { + byte[] overallMessage = new byte[] {(byte)0x90, 0x03, //fixed header + 0x00, 0x0A, //MSG ID + 0x01}; //QoS array + m_buff = Unpooled.buffer(overallMessage.length); + m_buff.writeBytes(overallMessage); + + //Exercise + m_msgdec.decode(null, m_buff, m_results); + + assertFalse(m_results.isEmpty()); + SubAckMessage message = (SubAckMessage)m_results.get(0); + assertNotNull(message); + assertEquals(0x0A, message.getMessageID().intValue()); + assertEquals(1, message.types().size()); + assertEquals(AbstractMessage.QOSType.LEAST_ONE, message.types().get(0)); + } + + private void initHeaderQos(ByteBuf buff, int messageID, AbstractMessage.QOSType... qoss) throws IllegalAccessException { + buff.clear().writeByte(AbstractMessage.SUBACK << 4). + writeBytes(Utils.encodeRemainingLength(2 + qoss.length)); + + buff.writeShort(messageID); + for (AbstractMessage.QOSType qos : qoss) { + buff.writeByte(qos.ordinal()); + } + } +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoderTest.java new file mode 100644 index 00000000..425fb78b --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/SubAckEncoderTest.java @@ -0,0 +1,62 @@ +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.SubAckMessage; +import static org.junit.Assert.assertEquals; +import org.junit.Before; +import org.junit.Test; + +/** + * + * @author andrea + */ +public class SubAckEncoderTest { + SubAckEncoder m_encoder = new SubAckEncoder(); + ChannelHandlerContext m_mockedContext; + ByteBuf m_out; + + @Before + public void setUp() { + //mock the ChannelHandlerContext to return an UnpooledAllocator + m_mockedContext = mockChannelHandler(); + m_out = Unpooled.buffer(); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncodeWithNoQoss() throws Exception { + SubAckMessage msg = new SubAckMessage(); + msg.setMessageID(123); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + } + + @Test + public void testEncodeWithMultipleQos() throws Exception { + SubAckMessage msg = new SubAckMessage(); + + int messageID = 0xAABB; + msg.setMessageID(messageID); + msg.addType(AbstractMessage.QOSType.MOST_ONE); + msg.addType(AbstractMessage.QOSType.LEAST_ONE); + msg.addType(AbstractMessage.QOSType.EXACTLY_ONCE); + + //Exercise + m_encoder.encode(m_mockedContext, msg, m_out); + + //Verify + assertEquals((byte) (AbstractMessage.SUBACK << 4 ), m_out.readByte()); //1 byte + assertEquals(5, m_out.readByte()); //remaining length + + //Variable part + assertEquals((byte)0xAA, m_out.readByte()); //MessageID MSB + assertEquals((byte)0xBB, m_out.readByte()); //MessageID LSB + assertEquals((byte)AbstractMessage.QOSType.MOST_ONE.ordinal(), m_out.readByte()); + assertEquals((byte)AbstractMessage.QOSType.LEAST_ONE.ordinal(), m_out.readByte()); + assertEquals((byte)AbstractMessage.QOSType.EXACTLY_ONCE.ordinal(), m_out.readByte()); + } +}