anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added Connect encoder part

Browse Source
This commit is contained in:
andrea
2013-09-11 15:42:00 +02:00
parent 557c682a5e
commit 3f44eaee39
5 changed files with 320 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoder.java
+77
View File
@@ -0,0 +1,77 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
/**
*
* @author andrea
*/
public class MQTTEncoder extends MessageToByteEncoder<AbstractMessage> {
@Override
protected void encode(ChannelHandlerContext chc, AbstractMessage msg, ByteBuf bb) throws Exception {
if (msg.getMessageType() == AbstractMessage.CONNECT) {
encodeConnect(chc, (ConnectMessage)msg, bb);
}
}
private void encodeConnect(ChannelHandlerContext chc, ConnectMessage message, ByteBuf out) {
ByteBuf staticHeaderBuff = chc.alloc().buffer(12);
staticHeaderBuff.writeBytes(Utils.encodeString("MQIsdp"));
//version
staticHeaderBuff.writeByte(0x03);
//connection flags and Strings
byte connectionFlags = 0;
if (message.isCleanSession()) {
connectionFlags |= 0x02;
}
if (message.isWillFlag()) {
connectionFlags |= 0x04;
}
connectionFlags |= ((message.getWillQos() & 0x03) << 3);
if (message.isWillRetain()) {
connectionFlags |= 0x020;
}
if (message.isPasswordFlag()) {
connectionFlags |= 0x040;
}
if (message.isUserFlag()) {
connectionFlags |= 0x080;
}
staticHeaderBuff.writeByte(connectionFlags);
//Keep alive timer
staticHeaderBuff.writeShort(message.getKeepAlive());
//Variable part
ByteBuf variableHeaderBuff = chc.alloc().buffer(12);
if (message.getClientID() != null) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getClientID()));
if (message.isWillFlag()) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillTopic()));
variableHeaderBuff.writeBytes(Utils.encodeString(message.getWillMessage()));
}
if (message.isUserFlag() && message.getUsername() != null) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getUsername()));
if (message.isPasswordFlag() && message.getPassword() != null) {
variableHeaderBuff.writeBytes(Utils.encodeString(message.getPassword()));
}
}
}
int variableHeaderSize = variableHeaderBuff.readableBytes();
ByteBuf buff = chc.alloc().buffer();
buff.writeByte(AbstractMessage.CONNECT << 4);
buff.writeBytes(Utils.encodeRemainingLength(12 + variableHeaderSize));
buff.writeBytes(staticHeaderBuff).writeBytes(variableHeaderBuff);
out.writeBytes(buff);
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java
+29
View File
@@ -2,6 +2,7 @@ package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.CorruptedFrameException;
import java.io.UnsupportedEncodingException;
import org.slf4j.LoggerFactory;
@@ -11,6 +12,8 @@ import org.slf4j.LoggerFactory;
*/
public class Utils {
public static final int MAX_LENGTH_LIMIT = 268435455;
static byte readMessageType(ByteBuf in) {
byte h1 = in.readByte();
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
@@ -60,6 +63,32 @@ public class Utils {
return value;
}
/**
* Encode the value in the format defined in specification as variable length
* array.
*
* @throws IllegalArgumentException if the value is not in the specification bounds
* [0..268435455].
*/
static ByteBuf encodeRemainingLength(int value) throws CorruptedFrameException {
if (value > MAX_LENGTH_LIMIT || value < 0) {
throw new CorruptedFrameException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value);
}
ByteBuf encoded = Unpooled.buffer(4);
byte digit;
do {
digit = (byte) (value % 128);
value = value / 128;
// if there are more digits to encode, set the top bit of this digit
if (value > 0) {
digit = (byte) (digit | 0x80);
}
encoded.writeByte(digit);
} while (value > 0);
return encoded;
}
/**
* Load a string from the given buffer, reading first the two bytes of len
* and then the UTF-8 bytes of the string.
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTEncoderTest.java
+133
View File
@@ -0,0 +1,133 @@
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.ConnectMessage;
import org.junit.Test;
import org.junit.Before;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.*;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
/**
*
* @author andrea
*/
public class MQTTEncoderTest {
MQTTEncoder m_encoder = new MQTTEncoder();
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 testBaseHeader() throws Exception {
ConnectMessage msg = new ConnectMessage();
msg.setWillRetain(true);
msg.setWillQos((byte)2);
msg.setWillFlag(false);
msg.setCleanSession(true);
msg.setKeepAlive(512);
//mock the ChannelHandlerContext to return an UnpooledAllocator
/*ChannelHandlerContext mockedContext = mock(ChannelHandlerContext.class);
ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT;
when(mockedContext.alloc()).thenReturn(allocator);*/
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x10, out.readByte()); //1 byte
assertEquals(12, out.readByte()); //remaining length
verifyString("MQIsdp", out);
assertEquals(0x03, out.readByte()); //protocol version
assertEquals(0x32, out.readByte()); //flags
assertEquals(2, out.readByte()); //keepAliveTimer msb
assertEquals(0, out.readByte()); //keepAliveTimer lsb
}
@Test
public void testCompleteHeader() throws Exception {
ConnectMessage msg = new ConnectMessage();
msg.setWillRetain(true);
msg.setWillQos((byte)2);
msg.setWillFlag(true);
msg.setCleanSession(true);
msg.setKeepAlive(512);
//variable part
msg.setClientID("ABCDEF");
msg.setWillTopic("Topic");
msg.setWillMessage("Message");
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x10, out.readByte()); //1 byte
assertEquals(36, out.readByte()); //remaining length
verifyString("MQIsdp", out);
assertEquals(0x03, out.readByte()); //protocol version
assertEquals(0x36, out.readByte()); //flags
assertEquals(2, out.readByte()); //keepAliveTimer msb
assertEquals(0, out.readByte()); //keepAliveTimer lsb
//Variable part
verifyString("ABCDEF", out);
verifyString("Topic", out);
verifyString("Message", out);
}
@Test
public void testCompleteHeaderWIthUser_password() throws Exception {
ConnectMessage msg = new ConnectMessage();
msg.setUserFlag(true);
msg.setPasswordFlag(true);
msg.setWillRetain(true);
msg.setWillQos((byte)2);
msg.setWillFlag(true);
msg.setCleanSession(true);
msg.setKeepAlive(512);
//variable part
msg.setClientID("ABCDEF");
msg.setWillTopic("Topic");
msg.setWillMessage("Message");
msg.setUsername("Pablo");
msg.setPassword("PBL");
ByteBuf out = Unpooled.buffer();
//Exercise
m_encoder.encode(m_mockedContext, msg, out);
//Verify
assertEquals(0x10, out.readByte()); //1 byte
assertEquals(48, out.readByte()); //remaining length
verifyString("MQIsdp", out);
assertEquals(0x03, out.readByte()); //protocol version
assertEquals((byte)0xF6, (byte)out.readByte()); //flags
assertEquals(2, out.readByte()); //keepAliveTimer msb
assertEquals(0, out.readByte()); //keepAliveTimer lsb
//Variable part
verifyString("ABCDEF", out);
verifyString("Topic", out);
verifyString("Message", out);
verifyString("Pablo", out);//username
verifyString("PBL", out);//password
}
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/TestUtils.java
+49
View File
@@ -0,0 +1,49 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.io.UnsupportedEncodingException;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
/**
* Common test methods
*
* @author andrea
*/
public class TestUtils {
/**
* Verify that the given bytes buffer of the given numBytes length is present
* in the buff starting from the current position.
*/
static void verifyBuff(int numBytes, byte[] bytes, ByteBuf buff) {
assertTrue(numBytes <= buff.readableBytes());
byte[] toCheck = new byte[numBytes];
buff.readBytes(toCheck);
for (int i = 0; i < numBytes; i++) {
assertEquals(bytes[i], toCheck[i]);
}
}
/**
* Verify the presence of the given string starting from the current position
* inside the buffer.
*/
static void verifyString(String str, ByteBuf buff) throws UnsupportedEncodingException {
ByteBuf tmpBuff = Unpooled.buffer(2);
byte[] raw = str.getBytes("UTF-8");
tmpBuff.writeShort(raw.length);
tmpBuff.writeBytes(raw);
int buffLen = raw.length + 2;
verifyByteBuf(tmpBuff, buff.slice(buff.readerIndex(), buffLen));
buff.skipBytes(buffLen);
}
static void verifyByteBuf(ByteBuf expected, ByteBuf found) {
assertEquals(expected, found);
}
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UtilsTest.java
+32 -1
View File
@@ -2,9 +2,11 @@ package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import static org.junit.Assert.assertEquals;
import io.netty.handler.codec.CorruptedFrameException;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.*;
import static org.dna.mqtt.moquette.parser.netty.TestUtils.*;
/**
*
@@ -47,4 +49,33 @@ public class UtilsTest {
m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, (byte)0x7F});
assertEquals(268435455, Utils.decodeRemainingLenght(m_buff));
}
@Test(expected=CorruptedFrameException.class)
public void testEncodeRemainingLength_invalid_upper() {
Utils.encodeRemainingLength(Utils.MAX_LENGTH_LIMIT + 1);
}
@Test(expected=CorruptedFrameException.class)
public void testEncodeRemainingLength_invalid_lower() {
Utils.encodeRemainingLength(-1);
}
@Test
public void testEncodeRemainingLenght() {
//1 byte length
verifyBuff(1, new byte[]{0}, Utils.encodeRemainingLength(0));
verifyBuff(1, new byte[]{0x7F}, Utils.encodeRemainingLength(127));
//2 byte length
verifyBuff(2, new byte[]{(byte)0x80, 0x01}, Utils.encodeRemainingLength(128));
verifyBuff(2, new byte[]{(byte)0xFF, 0x7F}, Utils.encodeRemainingLength(16383));
//3 byte length
verifyBuff(3, new byte[]{(byte)0x80, (byte)0x80, 0x01}, Utils.encodeRemainingLength(16384));
verifyBuff(3, new byte[]{(byte)0xFF, (byte)0xFF, 0x7F}, Utils.encodeRemainingLength(2097151));
//4 byte length
verifyBuff(4, new byte[]{(byte)0x80, (byte)0x80, (byte)0x80, 0x01}, Utils.encodeRemainingLength(2097152));
verifyBuff(4, new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, 0x7F}, Utils.encodeRemainingLength(268435455));
}
}