diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckDecoder.java new file mode 100644 index 00000000..6100e70a --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckDecoder.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.ConnAckMessage; + +/** + * + * @author andrea + */ +class ConnAckDecoder extends DemuxDecoder { + + @Override + void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { + in.resetReaderIndex(); + //Common decoding part + ConnAckMessage message = new ConnAckMessage(); + if (!decodeCommonHeader(message, in)) { + in.resetReaderIndex(); + return; + } + //skip reserved byte + in.skipBytes(1); + + //read return code + message.setReturnCode(in.readByte()); + out.add(message); + } + +} diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java new file mode 100644 index 00000000..e84abdcd --- /dev/null +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoder.java @@ -0,0 +1,22 @@ +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.ConnAckMessage; + +/** + * + * @author andrea + */ +class ConnAckEncoder extends DemuxEncoder{ + + @Override + protected void encode(ChannelHandlerContext chc, ConnAckMessage message, ByteBuf out) { + out.writeByte(AbstractMessage.CONNACK << 4); + out.writeBytes(Utils.encodeRemainingLength(2)); + out.writeByte(0); + out.writeByte(message.getReturnCode()); + } + +} 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 3021a397..86fb02c5 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 @@ -19,6 +19,7 @@ public class MQTTDecoder extends ByteToMessageDecoder { MQTTDecoder() { m_decoderMap.put(AbstractMessage.CONNECT, new ConnectDecoder()); + m_decoderMap.put(AbstractMessage.CONNACK, new ConnAckDecoder()); } @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 50f4062b..8b06003b 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 @@ -18,6 +18,7 @@ public class MQTTEncoder extends MessageToByteEncoder { MQTTEncoder() { m_encoderMap.put(AbstractMessage.CONNECT, new ConnectEncoder()); + m_encoderMap.put(AbstractMessage.CONNACK, new ConnAckEncoder()); } @Override diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnAckDecoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnAckDecoderTest.java new file mode 100644 index 00000000..8dd1915e --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnAckDecoderTest.java @@ -0,0 +1,53 @@ +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.ConnAckMessage; +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 ConnAckDecoderTest { + ByteBuf m_buff; + ConnAckDecoder m_msgdec; + + @Before + public void setUp() { + m_msgdec = new ConnAckDecoder(); + } + + @Test + public void testHeader() throws Exception { + m_buff = Unpooled.buffer(14); + initHeader(m_buff); + List results = new ArrayList(); + + //Excercise + m_msgdec.decode(null, m_buff, results); + + //Verify + assertFalse(results.isEmpty()); + ConnAckMessage message = (ConnAckMessage)results.get(0); + assertNotNull(message); + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, message.getReturnCode()); + assertEquals(AbstractMessage.CONNACK, message.getMessageType()); + } + + private void initHeader(ByteBuf buff) { + buff.clear().writeByte(AbstractMessage.CONNACK << 4).writeByte(2); + //reserved + buff.writeByte((byte)0); + //return code + buff.writeByte(ConnAckMessage.CONNECTION_ACCEPTED); + } + +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoderTest.java new file mode 100644 index 00000000..c223301d --- /dev/null +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnAckEncoderTest.java @@ -0,0 +1,48 @@ +package org.dna.mqtt.moquette.parser.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.Unpooled; +import io.netty.buffer.UnpooledByteBufAllocator; +import io.netty.channel.ChannelHandlerContext; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * + * @author andrea + */ +public class ConnAckEncoderTest { + + ConnAckEncoder m_encoder = new ConnAckEncoder(); + ChannelHandlerContext m_mockedContext; + + @Before + public void setUp() { + //mock the ChannelHandlerContext to return an UnpooledAllocator + m_mockedContext = mock(ChannelHandlerContext.class); + ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT; + when(m_mockedContext.alloc()).thenReturn(allocator); + } + + @Test + public void testHeaderEncode() throws Exception { + ConnAckMessage msg = new ConnAckMessage(); + + //Exercise + ByteBuf out = Unpooled.buffer(); + + //Exercise + m_encoder.encode(m_mockedContext, msg, out); + + //Verify + assertEquals(0x20, out.readByte()); //1 byte + assertEquals(0x02, out.readByte()); //2 byte, length + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, out.skipBytes(1).readByte()); + } + +} diff --git a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java index 8bbb3a3c..eccbc394 100644 --- a/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java +++ b/netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/ConnectEncoderTest.java @@ -8,7 +8,7 @@ import io.netty.channel.ChannelHandlerContext; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.junit.Test; import org.junit.Before; -import static org.junit.Assert.assertEquals; +import static org.junit.Assert.*; import static org.mockito.Mockito.*; import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;