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

Introduced raw decoder for CONNECT messages

Browse Source
This commit is contained in:
andrea
2013-09-11 11:06:53 +02:00
parent 2886aa0bab
commit 2ee84797ba
5 changed files with 424 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netty_parser/pom.xml
+6
View File
@@ -19,6 +19,12 @@
</properties>
<dependencies>
<dependency>
<groupId>org.dna.mqtt</groupId>
<artifactId>moquette-parser-commons</artifactId>
<version>0.2</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoder.java
+183
View File
@@ -0,0 +1,183 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.CorruptedFrameException;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
/**
*
* @author andrea
*/
public class MQTTDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
in.markReaderIndex();
if (!Utils.checkHeaderAvailability(in)) {
in.resetReaderIndex();
return;
}
in.resetReaderIndex();
byte messageType = Utils.readMessageType(in);
if (messageType == AbstractMessage.CONNECT) {
in.resetReaderIndex();
//Common decoding part
ConnectMessage message = new ConnectMessage();
if (!decodeCommonHeader(message, in)) {
in.resetReaderIndex();
return;
}
int remainingLength = message.getRemainingLength();
int start = in.readerIndex();
//Connect specific decoding part
//ProtocolName 8 bytes
if (in.readableBytes() < 12) {
in.resetReaderIndex();
return;
}
byte[] encProtoName = new byte[6];
in.skipBytes(2); //size, is 0x06
in.readBytes(encProtoName);
String protoName = new String(encProtoName, "UTF-8");
if (!"MQIsdp".equals(protoName)) {
in.resetReaderIndex();
throw new CorruptedFrameException("Invalid protoName: " + protoName);
}
message.setProtocolName(protoName);
//ProtocolVersion 1 byte (value 0x03)
message.setProcotolVersion(in.readByte());
//Connection flag
byte connFlags = in.readByte();
boolean cleanSession = ((connFlags & 0x02) >> 1) == 1 ? true : false;
boolean willFlag = ((connFlags & 0x04) >> 2) == 1 ? true : false;
byte willQos = (byte) ((connFlags & 0x18) >> 3);
if (willQos > 2) {
in.resetReaderIndex();
throw new CorruptedFrameException("Expected will QoS in range 0..2 but found: " + willQos);
}
boolean willRetain = ((connFlags & 0x20) >> 5) == 1 ? true : false;
boolean passwordFlag = ((connFlags & 0x40) >> 6) == 1 ? true : false;
boolean userFlag = ((connFlags & 0x80) >> 7) == 1 ? true : false;
//a password is true iff user is true.
if (!userFlag && passwordFlag) {
in.resetReaderIndex();
throw new CorruptedFrameException("Expected password flag to true if the user flag is true but was: " + passwordFlag);
}
message.setCleanSession(cleanSession);
message.setWillFlag(willFlag);
message.setWillQos(willQos);
message.setWillRetain(willRetain);
message.setPasswordFlag(passwordFlag);
message.setUserFlag(userFlag);
//Keep Alive timer 2 bytes
//int keepAlive = Utils.readWord(in);
int keepAlive = in.readShort();
message.setKeepAlive(keepAlive);
if (remainingLength == 12) {
out.add(message);
return;
}
//Decode the ClientID
String clientID = Utils.decodeString(in);
if (clientID == null) {
in.resetReaderIndex();
return;
}
message.setClientID(clientID);
//Decode willTopic
if (willFlag) {
String willTopic = Utils.decodeString(in);
if (willTopic == null) {
in.resetReaderIndex();
return;
}
message.setWillTopic(willTopic);
}
//Decode willMessage
if (willFlag) {
String willMessage = Utils.decodeString(in);
if (willMessage == null) {
in.resetReaderIndex();
return;
}
message.setWillMessage(willMessage);
}
//Compatibility check wieth v3.0, remaining length has precedence over
//the user and password flags
int readed = in.readerIndex() - start;
if (readed == remainingLength) {
out.add(message);
return;
}
//Decode username
if (userFlag) {
String userName = Utils.decodeString(in);
if (userName == null) {
in.resetReaderIndex();
return;
}
message.setUsername(userName);
}
readed = in.readerIndex() - start;
if (readed == remainingLength) {
out.add(message);
return;
}
//Decode password
if (passwordFlag) {
String password = Utils.decodeString(in);
if (password == null) {
in.resetReaderIndex();
return;
}
message.setPassword(password);
}
out.add(message);
return;
}
}
protected boolean decodeCommonHeader(AbstractMessage message, ByteBuf in) {
//Common decoding part
if (in.readableBytes() < 2) {
return false;
}
byte h1 = in.readByte();
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
boolean dupFlag = ((byte) ((h1 & 0x0008) >> 3) == 1);
byte qosLevel = (byte) ((h1 & 0x0006) >> 1);
boolean retainFlag = ((byte) (h1 & 0x0001) == 1);
int remainingLength = Utils.decodeRemainingLenght(in);
if (remainingLength == -1) {
return false;
}
message.setMessageType(messageType);
message.setDupFlag(dupFlag);
message.setQos(AbstractMessage.QOSType.values()[qosLevel]);
message.setRetainFlag(retainFlag);
message.setRemainingLength(remainingLength);
return true;
}
}
netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/Utils.java
+105
View File
@@ -0,0 +1,105 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.io.UnsupportedEncodingException;
import org.slf4j.LoggerFactory;
/**
*
* @author andrea
*/
public class Utils {
static byte readMessageType(ByteBuf in) {
byte h1 = in.readByte();
byte messageType = (byte) ((h1 & 0x00F0) >> 4);
return messageType;
}
static boolean checkHeaderAvailability(ByteBuf in) {
if (in.readableBytes() < 1) {
return false;
}
//byte h1 = in.get();
//byte messageType = (byte) ((h1 & 0x00F0) >> 4);
in.skipBytes(1); //skip the messageType byte
int remainingLength = Utils.decodeRemainingLenght(in);
if (remainingLength == -1) {
return false;
}
//check remaining length
if (in.readableBytes() < remainingLength) {
return false;
}
//return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK;
return true;
}
/**
* Decode the variable remaining length as defined in MQTT v3.1 specification
* (section 2.1).
*
* @return the decoded length or -1 if needed more data to decode the length field.
*/
static int decodeRemainingLenght(ByteBuf in) {
int multiplier = 1;
int value = 0;
byte digit;
do {
if (in.readableBytes() < 1) {
return -1;
}
digit = in.readByte();
value += (digit & 0x7F) * multiplier;
multiplier *= 128;
} while ((digit & 0x80) != 0);
return value;
}
/**
* Load a string from the given buffer, reading first the two bytes of len
* and then the UTF-8 bytes of the string.
*
* @return the decoded string or null if NEED_DATA
*/
static String decodeString(ByteBuf in) throws UnsupportedEncodingException {
if (in.readableBytes() < 2) {
return null;
}
//int strLen = Utils.readWord(in);
int strLen = in.readShort();
if (in.readableBytes() < strLen) {
return null;
}
byte[] strRaw = new byte[strLen];
in.readBytes(strRaw);
return new String(strRaw, "UTF-8");
}
/**
* Return the IoBuffer with string encoded as MSB, LSB and UTF-8 encoded
* string content.
*/
static ByteBuf encodeString(String str) {
ByteBuf out = Unpooled.buffer(2);
byte[] raw;
try {
raw = str.getBytes("UTF-8");
//NB every Java platform has got UTF-8 encoding by default, so this
//exception are never raised.
} catch (UnsupportedEncodingException ex) {
LoggerFactory.getLogger(Utils.class).error(null, ex);
return null;
}
//Utils.writeWord(out, raw.length);
out.writeShort(raw.length);
out.writeBytes(raw);
return out;
}
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/MQTTDecoderTest.java
+80
View File
@@ -0,0 +1,80 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import org.junit.Before;
import org.junit.Test;
public class MQTTDecoderTest {
ByteBuf m_buff;
MQTTDecoder m_msgdec;
@Before
public void setUp() {
m_msgdec = new MQTTDecoder();
}
@Test
public void testBaseHeader() throws Exception {
m_buff = Unpooled.buffer(14);
initBaseHeader(m_buff);
//m_buff.flip();
List<Object> results = new ArrayList<Object >();
//Excercise
m_msgdec.decode(null, m_buff, results);
//Verify
assertFalse(results.isEmpty());
verifyBaseHeader((ConnectMessage)results.get(0));
}
private void initBaseHeader(ByteBuf buff) throws UnsupportedEncodingException {
initHeader(buff, (byte)0x0C);
}
private void initHeader(ByteBuf buff, byte remaingLength) throws UnsupportedEncodingException {
buff.clear().writeByte((byte)(AbstractMessage.CONNECT << 4)).writeByte(remaingLength);
//Proto name
encodeString(buff, "MQIsdp");
//version
buff.writeByte(3);
//conn flags
buff.writeByte(0xCE);
//keepAlive
buff.writeBytes(new byte[]{(byte)0, (byte) 0x0A});
}
/**
* Encode and insert the given string into the given buff
*/
private ByteBuf encodeString(ByteBuf buff, String str) throws UnsupportedEncodingException {
buff.writeBytes(Utils.encodeString(str));
return buff;
}
private void verifyBaseHeader(ConnectMessage connMessage) {
assertNotNull(connMessage);
assertEquals("MQIsdp", connMessage.getProtocolName());
assertEquals(3, connMessage.getProcotolVersion());
assertTrue(connMessage.isUserFlag());
assertTrue(connMessage.isPasswordFlag());
assertTrue(connMessage.isCleanSession());
assertEquals(10, connMessage.getKeepAlive());
assertTrue(connMessage.isWillFlag());
assertFalse(connMessage.isWillRetain());
assertEquals(1, connMessage.getWillQos());
}
}
netty_parser/src/test/java/org/dna/mqtt/moquette/parser/netty/UtilsTest.java
+50
View File
@@ -0,0 +1,50 @@
package org.dna.mqtt.moquette.parser.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import static org.junit.Assert.assertEquals;
import org.junit.Before;
import org.junit.Test;
/**
*
* @author andrea
*/
/*@RunWith(Suite.class)
@Suite.SuiteClasses({})*/
public class UtilsTest {
ByteBuf m_buff;
@Before
public void setUp() {
m_buff = Unpooled.buffer(4);
}
@Test
public void testDecodeRemainingLength() {
//1 byte length
m_buff.writeByte(0x0);
assertEquals(0, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeByte(0x007F);
assertEquals(127, Utils.decodeRemainingLenght(m_buff));
//2 byte length
m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x01});
assertEquals(128, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0x7F});
assertEquals(16383, Utils.decodeRemainingLenght(m_buff));
//3 byte length
m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x80, (byte)0x01});
assertEquals(16384, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0x7F});
assertEquals(2097151, Utils.decodeRemainingLenght(m_buff));
//4 byte length
m_buff.clear().writeBytes(new byte[]{(byte)0x80, (byte)0x80, (byte)0x80, (byte)0x01});
assertEquals(2097152, Utils.decodeRemainingLenght(m_buff));
m_buff.clear().writeBytes(new byte[]{(byte)0xFF, (byte)0xFF, (byte)0xFF, (byte)0x7F});
assertEquals(268435455, Utils.decodeRemainingLenght(m_buff));
}
}