Added encode/decode for ConnAck message
This commit is contained in:
@@ -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<Object> 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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<ConnAckMessage>{
|
||||
|
||||
@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());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -18,6 +18,7 @@ public class MQTTEncoder extends MessageToByteEncoder<AbstractMessage> {
|
||||
|
||||
MQTTEncoder() {
|
||||
m_encoderMap.put(AbstractMessage.CONNECT, new ConnectEncoder());
|
||||
m_encoderMap.put(AbstractMessage.CONNACK, new ConnAckEncoder());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -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<Object> results = new ArrayList<Object >();
|
||||
|
||||
//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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
+1
-1
@@ -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.*;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user