commit 422fbc4d2c644844d4886afd55c912626d1c4054 Author: selva.andre Date: Tue Jan 3 18:25:50 2012 +0100 First import of moquette proto parser diff --git a/docs/MQTT_V3.1_Protocol_Specific.pdf b/docs/MQTT_V3.1_Protocol_Specific.pdf new file mode 100644 index 00000000..932ce4a3 Binary files /dev/null and b/docs/MQTT_V3.1_Protocol_Specific.pdf differ diff --git a/pom.xml b/pom.xml new file mode 100644 index 00000000..12943139 --- /dev/null +++ b/pom.xml @@ -0,0 +1,56 @@ + + + 4.0.0 + + org.dna.mqtt + moquette + 0.1-SNAPSHOT + jar + + moquette + http://maven.apache.org + + + UTF-8 + + + + + org.apache.mina + mina-core + 2.0.4 + + + + junit + junit + 4.8.2 + test + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + 1.5 + 1.5 + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + org/dna/mqtt/moquette/proto/TestUtils.java + org/dna/mqtt/moquette/proto/TestUtils$* + + + + + + diff --git a/src/main/java/org/dna/mqtt/moquette/proto/ConnAckDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/ConnAckDecoder.java new file mode 100644 index 00000000..4c62ff50 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/ConnAckDecoder.java @@ -0,0 +1,35 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; + +/** + * + * @author andrea + */ +public class ConnAckDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.CONNACK, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + ConnAckMessage message = new ConnAckMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + //skip reserved byte + in.skip(1); + + //read return code + message.setReturnCode(in.get()); + out.write(message); + return OK; + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/ConnAckEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/ConnAckEncoder.java new file mode 100644 index 00000000..7c2d4028 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/ConnAckEncoder.java @@ -0,0 +1,24 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; + +/** + * + * @author andrea + */ +public class ConnAckEncoder implements MessageEncoder { + + public void encode(IoSession session, ConnAckMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(4); + buff.put((byte) (AbstractMessage.CONNACK << 4)); + buff.put(Utils.encodeRemainingLength(2)); + buff.put((byte) 0); + buff.put(message.getReturnCode()).flip(); + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/ConnectDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/ConnectDecoder.java new file mode 100644 index 00000000..408baffb --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/ConnectDecoder.java @@ -0,0 +1,138 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; + +/** + * Decoder for the CONNECT message + * + * @author andrea + */ +public class ConnectDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.CONNECT, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + ConnectMessage message = new ConnectMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + int remainingLength = message.getRemainingLength(); + int start = in.position(); + + //Connect specific decoding part + //ProtocolName 8 bytes + if (in.remaining() < 12) { + return NEED_DATA; + } + byte[] encProtoName = new byte[6]; + in.skip(2); //size, is 0x06 + in.get(encProtoName); + String protoName = new String(encProtoName, "UTF-8"); + if (!"MQIsdp".equals(protoName)) { + return NOT_OK; + } + message.setProtocolName(protoName); + + //ProtocolVersion 1 byte (value 0x03) + message.setProcotolVersion(in.get()); + + //Connection flag + byte connFlags = in.get(); + 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) { + return NOT_OK; //admitted values for willqos are 0, 1, 2 + } + 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) { + return NOT_OK; + } + 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); + message.setKeepAlive(keepAlive); + + if (remainingLength == 12) { + out.write(message); + return OK; + } + + //Decode the ClientID + String clientID = Utils.decodeString(in); + if (clientID == null) { + return NEED_DATA; + } + message.setClientID(clientID); + + //Decode willTopic + if (willFlag) { + String willTopic = Utils.decodeString(in); + if (willTopic == null) { + return NEED_DATA; + } + message.setWillTopic(willTopic); + } + + //Decode willMessage + if (willFlag) { + String willMessage = Utils.decodeString(in); + if (willMessage == null) { + return NEED_DATA; + } + message.setWillMessage(willMessage); + } + + //Compatibility check wieth v3.0, remaining length has precedence over + //the user and password flags + int readed = in.position() - start; + if (readed == remainingLength) { + out.write(message); + return OK; + } + + //Decode username + if (userFlag) { + String userName = Utils.decodeString(in); + if (userName == null) { + return NEED_DATA; + } + message.setUsername(userName); + } + + readed = in.position() - start; + if (readed == remainingLength) { + out.write(message); + return OK; + } + + //Decode password + if (passwordFlag) { + String password = Utils.decodeString(in); + if (password == null) { + return NEED_DATA; + } + message.setPassword(password); + } + + out.write(message); + return OK; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/ConnectEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/ConnectEncoder.java new file mode 100644 index 00000000..0034a303 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/ConnectEncoder.java @@ -0,0 +1,69 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; + +/** + * + * @author andrea + */ +public class ConnectEncoder implements MessageEncoder { + + public void encode(IoSession session, ConnectMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer staticHeaderBuff = IoBuffer.allocate(12); + staticHeaderBuff.put(Utils.encodeString("MQIsdp")); + + //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.put(connectionFlags); + + //Keep alive timer + Utils.writeWord(staticHeaderBuff, message.getKeepAlive()); + staticHeaderBuff.flip(); + + //Variable part + IoBuffer variableHeaderBuff = IoBuffer.allocate(12).setAutoExpand(true); + if (message.getClientID() != null) { + variableHeaderBuff.put(Utils.encodeString(message.getClientID())); + if (message.isWillFlag()) { + variableHeaderBuff.put(Utils.encodeString(message.getWillTopic())); + variableHeaderBuff.put(Utils.encodeString(message.getWillMessage())); + } + if (message.isUserFlag() && message.getUsername() != null) { + variableHeaderBuff.put(Utils.encodeString(message.getUsername())); + if (message.isPasswordFlag() && message.getPassword() != null) { + variableHeaderBuff.put(Utils.encodeString(message.getPassword())); + } + } + } + variableHeaderBuff.flip(); + + int variableHeaderSize = variableHeaderBuff.remaining(); + IoBuffer buff = IoBuffer.allocate(14 + variableHeaderSize); + buff.put((byte) (AbstractMessage.CONNECT << 4)); + buff.put(Utils.encodeRemainingLength(12 + variableHeaderSize)); + buff.put(staticHeaderBuff).put(variableHeaderBuff).flip(); + + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/DisconnectDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/DisconnectDecoder.java new file mode 100644 index 00000000..eac4904f --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/DisconnectDecoder.java @@ -0,0 +1,29 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.DisconnectMessage; + +/** + * + * @author andrea + */ +public class DisconnectDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.DISCONNECT, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + DisconnectMessage message = new DisconnectMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + out.write(message); + return OK; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/DisconnectEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/DisconnectEncoder.java new file mode 100644 index 00000000..0c25f531 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/DisconnectEncoder.java @@ -0,0 +1,22 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.DisconnectMessage; + +/** + * + * @author andrea + */ +public class DisconnectEncoder implements MessageEncoder { + + public void encode(IoSession session, DisconnectMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(2); + buff.put((byte) (AbstractMessage.DISCONNECT << 4)).put((byte)0).flip(); + out.write(buff); + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/MessageIDDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/MessageIDDecoder.java new file mode 100644 index 00000000..199421fd --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/MessageIDDecoder.java @@ -0,0 +1,28 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; + +/** + * + * @author andrea + */ +public abstract class MessageIDDecoder extends MqttDecoder { + protected abstract MessageIDMessage createMessage(); + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + MessageIDMessage message = createMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + + //read messageIDs + message.setMessageID(Utils.readWord(in)); + out.write(message); + return OK; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/MqttDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/MqttDecoder.java new file mode 100644 index 00000000..c28a65bf --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/MqttDecoder.java @@ -0,0 +1,37 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoderAdapter; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; + +/** + * Contains common code for all mqtt decoders. + * + * @author andrea + */ +public abstract class MqttDecoder extends MessageDecoderAdapter { + + protected MessageDecoderResult decodeCommonHeader(AbstractMessage message, IoBuffer in) { + //Common decoding part + if (in.remaining() < 2) { + return NEED_DATA; + } + byte h1 = in.get(); + byte messageType = (byte) ((h1 & 0x0010) >> 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 NEED_DATA; + } + + message.setMessageType(messageType); + message.setDupFlag(dupFlag); + message.setQos(AbstractMessage.QOSType.values()[qosLevel]); + message.setRetainFlag(retainFlag); + message.setRemainingLength(remainingLength); + return OK; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PingReqDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PingReqDecoder.java new file mode 100644 index 00000000..8584424d --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PingReqDecoder.java @@ -0,0 +1,29 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PingReqMessage; + +/** + * + * @author andrea + */ +public class PingReqDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PINGREQ, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + PingReqMessage message = new PingReqMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + out.write(message); + return OK; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PingReqEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PingReqEncoder.java new file mode 100644 index 00000000..b5f70862 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PingReqEncoder.java @@ -0,0 +1,21 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PingReqMessage; + +/** + * + * @author andrea + */ +public class PingReqEncoder implements MessageEncoder { + + public void encode(IoSession session, PingReqMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(2); + buff.put((byte) (AbstractMessage.PINGREQ << 4)).put((byte)0).flip(); + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PingRespDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PingRespDecoder.java new file mode 100644 index 00000000..bc0bef10 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PingRespDecoder.java @@ -0,0 +1,33 @@ +/* + * To change this template, choose Tools | Templates + * and open the template in the editor. + */ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PingRespMessage; + +/** + * + * @author andrea + */ +public class PingRespDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PINGRESP, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + PingRespMessage message = new PingRespMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + out.write(message); + return OK; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PingRespEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PingRespEncoder.java new file mode 100644 index 00000000..3afedf32 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PingRespEncoder.java @@ -0,0 +1,21 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PingRespMessage; + +/** + * + * @author andrea + */ +public class PingRespEncoder implements MessageEncoder { + + public void encode(IoSession session, PingRespMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(2); + buff.put((byte) (AbstractMessage.PINGRESP << 4)).put((byte)0).flip(); + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubAckDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubAckDecoder.java new file mode 100644 index 00000000..97e2c16c --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubAckDecoder.java @@ -0,0 +1,25 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; + +/** + * + * @author andrea + */ +public class PubAckDecoder extends MessageIDDecoder { + + @Override + protected MessageIDMessage createMessage() { + return new PubAckMessage(); + } + + @Override + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PUBACK, in); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubAckEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubAckEncoder.java new file mode 100644 index 00000000..1a2dcbfb --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubAckEncoder.java @@ -0,0 +1,23 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; + +/** + * + * @author andrea + */ +public class PubAckEncoder implements MessageEncoder { + public void encode(IoSession session, PubAckMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(4); + buff.put((byte) (AbstractMessage.PUBACK << 4)); + buff.put(Utils.encodeRemainingLength(2)); + Utils.writeWord(buff, message.getMessageID()); + buff.flip(); + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubCompDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubCompDecoder.java new file mode 100644 index 00000000..5c3d53d6 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubCompDecoder.java @@ -0,0 +1,24 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; + +/** + * + * @author andrea + */ +public class PubCompDecoder extends MessageIDDecoder { + + @Override + protected MessageIDMessage createMessage() { + return new PubCompMessage(); + } + + @Override + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PUBCOMP, in); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubCompEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubCompEncoder.java new file mode 100644 index 00000000..a368cff8 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubCompEncoder.java @@ -0,0 +1,22 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; + +/** + * + * @author andrea + */ +public class PubCompEncoder implements MessageEncoder { + public void encode(IoSession session, PubCompMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(4); + buff.put((byte) (AbstractMessage.PUBCOMP << 4)); + buff.put(Utils.encodeRemainingLength(2)); + Utils.writeWord(buff, message.getMessageID()); + buff.flip(); + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubCompMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/PubCompMessage.java new file mode 100644 index 00000000..1b5f0e2e --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubCompMessage.java @@ -0,0 +1,11 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; + +/** + * + * @author andrea + */ +public class PubCompMessage extends MessageIDMessage { +} + diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubRecDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubRecDecoder.java new file mode 100644 index 00000000..ad4d9b64 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubRecDecoder.java @@ -0,0 +1,25 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; +import org.dna.mqtt.moquette.proto.messages.PubRecMessage; + +/** + * + * @author andrea + */ +public class PubRecDecoder extends MessageIDDecoder { + + @Override + protected MessageIDMessage createMessage() { + return new PubRecMessage(); + } + + @Override + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PUBREC, in); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubRecEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubRecEncoder.java new file mode 100644 index 00000000..ba21a527 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubRecEncoder.java @@ -0,0 +1,23 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PubRecMessage; + +/** + * + * @author andrea + */ +public class PubRecEncoder implements MessageEncoder { + public void encode(IoSession session, PubRecMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(4); + buff.put((byte) (AbstractMessage.PUBREC << 4)); + buff.put(Utils.encodeRemainingLength(2)); + Utils.writeWord(buff, message.getMessageID()); + buff.flip(); + out.write(buff); + } +} \ No newline at end of file diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubRelDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubRelDecoder.java new file mode 100644 index 00000000..c2cb7d05 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubRelDecoder.java @@ -0,0 +1,26 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; +import org.dna.mqtt.moquette.proto.messages.PubRelMessage; + +/** + * + * @author andrea + */ +public class PubRelDecoder extends MessageIDDecoder { + + @Override + protected MessageIDMessage createMessage() { + return new PubRelMessage(); + } + + @Override + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PUBREL, in); + } +} + diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PubRelEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PubRelEncoder.java new file mode 100644 index 00000000..a314f253 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PubRelEncoder.java @@ -0,0 +1,23 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PubRelMessage; + +/** + * + * @author andrea + */ +public class PubRelEncoder implements MessageEncoder { + public void encode(IoSession session, PubRelMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(4); + buff.put((byte) (AbstractMessage.PUBREL << 4)); + buff.put(Utils.encodeRemainingLength(2)); + Utils.writeWord(buff, message.getMessageID()); + buff.flip(); + out.write(buff); + } +} \ No newline at end of file diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java new file mode 100644 index 00000000..c8ec8b42 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PublishDecoder.java @@ -0,0 +1,57 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; + +/** + * + * @author andrea + */ +public class PublishDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.PUBLISH, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + int startPos = in.position(); + + //Common decoding part + PublishMessage message = new PublishMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + int remainingLength = message.getRemainingLength(); + + //Topic name + String topic = Utils.decodeString(in); + if (topic == null) { + return NEED_DATA; + } + message.setTopicName(topic); + + if (message.getQos() == QOSType.LEAST_ONE || + message.getQos() == QOSType.EXACTLY_ONCE) { + message.setMessageID(Utils.readWord(in)); + } + int stopPos = in.position(); + + //read the payload + int payloadSize = remainingLength - (stopPos - startPos - 2); + byte[] b = new byte[payloadSize]; + if (in.remaining() < payloadSize) { + return NEED_DATA; + } + in.get(b); + message.setPayload(b); + + out.write(message); + return OK; + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java new file mode 100644 index 00000000..07df7499 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/PublishEncoder.java @@ -0,0 +1,48 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; + +/** + * + * @author andrea + */ +public class PublishEncoder implements MessageEncoder { + + public void encode(IoSession session, PublishMessage message, ProtocolEncoderOutput out) throws Exception { + if (message.getQos() == QOSType.RESERVED) { + throw new IllegalArgumentException("Found a message with RESERVED Qos"); + } + if (message.getTopicName() == null || message.getTopicName().isEmpty()) { + throw new IllegalArgumentException("Found a message with empty or null topic name"); + } + + IoBuffer variableHeaderBuff = IoBuffer.allocate(2).setAutoExpand(true); + variableHeaderBuff.put(Utils.encodeString(message.getTopicName())); + if (message.getQos() == QOSType.LEAST_ONE || + message.getQos() == QOSType.EXACTLY_ONCE ) { + if (message.getMessageID() == null) { + throw new IllegalArgumentException("Found a message with QOS 1 or 2 and not MessageID setted"); + } + Utils.writeWord(variableHeaderBuff, message.getMessageID()); + } + variableHeaderBuff.put(message.getPayload()); + variableHeaderBuff.flip(); + int variableHeaderSize = variableHeaderBuff.remaining(); + + byte flags = Utils.encodeFlags(message); + + IoBuffer buff = IoBuffer.allocate(4 + variableHeaderSize); + buff.put((byte) (AbstractMessage.PUBLISH << 4 | flags)); + buff.put(Utils.encodeRemainingLength(4 + variableHeaderSize)); + buff.put(variableHeaderBuff).flip(); + + out.write(buff); + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/SubAckDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/SubAckDecoder.java new file mode 100644 index 00000000..d9b84986 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/SubAckDecoder.java @@ -0,0 +1,46 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; + +/** + * QoS, DUP and RETAIN are NOT used. + * + * @author andrea + */ +public class SubAckDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.SUBACK, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + SubAckMessage message = new SubAckMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + int remainingLength = message.getRemainingLength(); + + //MessageID + message.setMessageID(Utils.readWord(in)); + + //Qos array + if (in.remaining() < remainingLength) { + return NEED_DATA; + } + for (int i = 0; i < remainingLength; i++) { + byte qos = in.get(); + message.addType(QOSType.values()[qos]); + } + + out.write(message); + return OK; + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/SubAckEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/SubAckEncoder.java new file mode 100644 index 00000000..7fb6a921 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/SubAckEncoder.java @@ -0,0 +1,39 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; + +/** + * + * @author andrea + */ +public class SubAckEncoder implements MessageEncoder { + + public void encode(IoSession session, SubAckMessage message, ProtocolEncoderOutput out) throws Exception { + if (message.types().isEmpty()) { + throw new IllegalArgumentException("Found a suback message with empty topics"); + } + + IoBuffer variableHeaderBuff = IoBuffer.allocate(4).setAutoExpand(true); + Utils.writeWord(variableHeaderBuff, message.getMessageID()); + for (AbstractMessage.QOSType c : message.types()) { + byte ord = (byte) c.ordinal(); + variableHeaderBuff.put((byte)c.ordinal()); + } + + variableHeaderBuff.flip(); + int variableHeaderSize = variableHeaderBuff.remaining(); + IoBuffer buff = IoBuffer.allocate(4 + variableHeaderSize); + + buff.put((byte) (AbstractMessage.SUBACK << 4 )); + buff.put(Utils.encodeRemainingLength(variableHeaderSize)); + buff.put(variableHeaderBuff).flip(); + + out.write(buff); + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/SubscribeDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/SubscribeDecoder.java new file mode 100644 index 00000000..536151cb --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/SubscribeDecoder.java @@ -0,0 +1,56 @@ +package org.dna.mqtt.moquette.proto; + +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; + +/** + * + * @author andrea + */ +public class SubscribeDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.SUBSCRIBE, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + SubscribeMessage message = new SubscribeMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + + //check qos level + if (message.getQos() != QOSType.LEAST_ONE) { + return NOT_OK; + } + + int start = in.position(); + //read messageIDs + message.setMessageID(Utils.readWord(in)); + int readed = in.position() - start; + while (readed < message.getRemainingLength()) { + decodeSubscription(in, message); + readed = in.position() - start; + } + + out.write(message); + return OK; + } + + /** + * Populate the message with couple of Qos, topic + */ + private void decodeSubscription(IoBuffer in, SubscribeMessage message) throws UnsupportedEncodingException { + String topic = Utils.decodeString(in); + byte qos = (byte)(in.get() & 0x03); + message.addSubscription(new SubscribeMessage.Couple(qos, topic)); + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/SubscribeEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/SubscribeEncoder.java new file mode 100644 index 00000000..d5e5b0c9 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/SubscribeEncoder.java @@ -0,0 +1,45 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage.Couple; + +/** + * + * @author andrea + */ +public class SubscribeEncoder implements MessageEncoder { + + public void encode(IoSession session, SubscribeMessage message, ProtocolEncoderOutput out) throws Exception { + if (message.subscriptions().isEmpty()) { + throw new IllegalArgumentException("Found a subscribe message with empty topics"); + } + + if (message.getQos() != QOSType.LEAST_ONE) { + throw new IllegalArgumentException("Expected a message with QOS 1, found " + message.getQos()); + } + + IoBuffer variableHeaderBuff = IoBuffer.allocate(4).setAutoExpand(true); + Utils.writeWord(variableHeaderBuff, message.getMessageID()); + for (Couple c : message.subscriptions()) { + variableHeaderBuff.put(Utils.encodeString(c.getTopic())); + variableHeaderBuff.put(c.getQos()); + } + + variableHeaderBuff.flip(); + int variableHeaderSize = variableHeaderBuff.remaining(); + byte flags = Utils.encodeFlags(message); + IoBuffer buff = IoBuffer.allocate(4 + variableHeaderSize); + + buff.put((byte) (AbstractMessage.SUBSCRIBE << 4 | flags)); + buff.put(Utils.encodeRemainingLength(4 + variableHeaderSize)); + buff.put(variableHeaderBuff).flip(); + + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/UnsubAckDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/UnsubAckDecoder.java new file mode 100644 index 00000000..8f91bae4 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/UnsubAckDecoder.java @@ -0,0 +1,26 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; + +/** + * + * @author andrea + */ +public class UnsubAckDecoder extends MessageIDDecoder { + + @Override + protected MessageIDMessage createMessage() { + return new UnsubAckMessage(); + } + + @Override + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.UNSUBACK, in); + } +} + diff --git a/src/main/java/org/dna/mqtt/moquette/proto/UnsubAckEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/UnsubAckEncoder.java new file mode 100644 index 00000000..0530221e --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/UnsubAckEncoder.java @@ -0,0 +1,23 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; + +/** + * + * @author andrea + */ +public class UnsubAckEncoder implements MessageEncoder { + public void encode(IoSession session, UnsubAckMessage message, ProtocolEncoderOutput out) throws Exception { + IoBuffer buff = IoBuffer.allocate(4); + buff.put((byte) (AbstractMessage.UNSUBACK << 4)); + buff.put(Utils.encodeRemainingLength(2)); + Utils.writeWord(buff, message.getMessageID()); + buff.flip(); + out.write(buff); + } +} \ No newline at end of file diff --git a/src/main/java/org/dna/mqtt/moquette/proto/UnsubscribeDecoder.java b/src/main/java/org/dna/mqtt/moquette/proto/UnsubscribeDecoder.java new file mode 100644 index 00000000..7cd85df2 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/UnsubscribeDecoder.java @@ -0,0 +1,48 @@ +package org.dna.mqtt.moquette.proto; + +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; + +/** + * + * @author andrea + */ +public class UnsubscribeDecoder extends MqttDecoder { + + public MessageDecoderResult decodable(IoSession session, IoBuffer in) { + return Utils.checkDecodable(AbstractMessage.UNSUBSCRIBE, in); + } + + public MessageDecoderResult decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception { + //Common decoding part + UnsubscribeMessage message = new UnsubscribeMessage(); + if (decodeCommonHeader(message, in) == NEED_DATA) { + return NEED_DATA; + } + + //check qos level + if (message.getQos() != QOSType.LEAST_ONE) { + return NOT_OK; + } + + int start = in.position(); + //read messageIDs + message.setMessageID(Utils.readWord(in)); + int readed = in.position() - start; + while (readed < message.getRemainingLength()) { + message.addTopic(Utils.decodeString(in)); + readed = in.position() - start; + } + + out.write(message); + return OK; + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/UnsubscribeEncoder.java b/src/main/java/org/dna/mqtt/moquette/proto/UnsubscribeEncoder.java new file mode 100644 index 00000000..a57f979f --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/UnsubscribeEncoder.java @@ -0,0 +1,43 @@ +package org.dna.mqtt.moquette.proto; + +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import org.apache.mina.filter.codec.demux.MessageEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; + +/** + * + * @author andrea + */ +public class UnsubscribeEncoder implements MessageEncoder { + + public void encode(IoSession session, UnsubscribeMessage message, ProtocolEncoderOutput out) throws Exception { + if (message.topics().isEmpty()) { + throw new IllegalArgumentException("Found an unsubscribe message with empty topics"); + } + + if (message.getQos() != QOSType.LEAST_ONE) { + throw new IllegalArgumentException("Expected a message with QOS 1, found " + message.getQos()); + } + + IoBuffer variableHeaderBuff = IoBuffer.allocate(4).setAutoExpand(true); + Utils.writeWord(variableHeaderBuff, message.getMessageID()); + for (String topic : message.topics()) { + variableHeaderBuff.put(Utils.encodeString(topic)); + } + + variableHeaderBuff.flip(); + int variableHeaderSize = variableHeaderBuff.remaining(); + byte flags = Utils.encodeFlags(message); + IoBuffer buff = IoBuffer.allocate(4 + variableHeaderSize); + + buff.put((byte) (AbstractMessage.UNSUBSCRIBE << 4 | flags)); + buff.put(Utils.encodeRemainingLength(4 + variableHeaderSize)); + buff.put(variableHeaderBuff).flip(); + + out.write(buff); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/Utils.java b/src/main/java/org/dna/mqtt/moquette/proto/Utils.java new file mode 100644 index 00000000..d98b5f5d --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/Utils.java @@ -0,0 +1,144 @@ +package org.dna.mqtt.moquette.proto; + +import java.io.UnsupportedEncodingException; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; + +/** + * Common utils methodd used in codecs. + * + * @author andrea + */ +public class Utils { + + public static final int MAX_LENGTH_LIMIT = 268435455; + + /** + * Read 2 bytes from in buffer first MSB, and then LSB returning as int. + */ + static int readWord(IoBuffer in) { + int msb = in.get() & 0x00FF; //remove sign extension due to casting + int lsb = in.get() & 0x00FF; + msb = (msb << 8) | lsb ; + return msb; + } + + /** + * Writes as 2 bytes the int value into buffer first MSB, and then LSB. + */ + static void writeWord(IoBuffer out, int value) { + out.put((byte) ((value & 0xFF00) >> 8)); //msb + out.put((byte) (value & 0x00FF)); //lsb + } + + /** + * Decode the variable remaining lenght as defined in MQTT v3.1 specification + * (section 2.1). + * + * @return the decoded length or -1 if needed more data to decode. + */ + static int decodeRemainingLenght(IoBuffer in) { + int multiplier = 1; + int value = 0; + byte digit; + do { + if (in.remaining() < 1) { + return -1; + } + digit = in.get(); + value += (digit & 0x7F) * multiplier; + multiplier *= 128; + } while ((digit & 0x80) != 0); + 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 IoBuffer encodeRemainingLength(int value) throws IllegalAccessException { + if (value > MAX_LENGTH_LIMIT || value < 0) { + throw new IllegalAccessException("Value should in range 0.." + MAX_LENGTH_LIMIT + " found " + value); + } + + IoBuffer encoded = IoBuffer.allocate(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.put(digit); + } while (value > 0); + encoded.flip(); + return encoded; + } + + static MessageDecoderResult checkDecodable(byte type, IoBuffer in) { + if (in.remaining() < 1) { + return MessageDecoderResult.NEED_DATA; + } + byte h1 = in.get(); + byte messageType = (byte) ((h1 & 0x00F0) >> 4); + return messageType == type ? MessageDecoderResult.OK : MessageDecoderResult.NOT_OK; + } + + /** + * Return the IoBuffer with string ancoded as MSB, LSB and UTF-8 encoded + * string content. + */ + static IoBuffer encodeString(String str) { + IoBuffer out = IoBuffer.allocate(2).setAutoExpand(true); + 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) { + Logger.getLogger(ConnectEncoder.class.getName()).log(Level.SEVERE, null, ex); + return null; + } + Utils.writeWord(out, raw.length); + out.put(raw).flip(); + return out; + } + + + /** + * 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(IoBuffer in) throws UnsupportedEncodingException { + if (in.remaining() < 2) { + return null; + } + int strLen = Utils.readWord(in); + if (in.remaining() < strLen) { + return null; + } + byte[] strRaw = new byte[strLen]; + in.get(strRaw); + + return new String(strRaw, "UTF-8"); + } + + static byte encodeFlags(AbstractMessage message) { + byte flags = 0; + if (message.isDupFlag()) { + flags |= 0x08; + } + + flags |= ((message.getQos().ordinal() & 0x03) << 1); + return flags; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java new file mode 100644 index 00000000..8e11b30e --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java @@ -0,0 +1,81 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * Basic abstract message for all MQTT protocol messages. + * + * @author andrea + */ +public abstract class AbstractMessage { + + public static final byte CONNECT = 1; // Client request to connect to Server + public static final byte CONNACK = 2; // Connect Acknowledgment + public static final byte PUBLISH = 3; // Publish message + public static final byte PUBACK = 4; // Publish Acknowledgment + public static final byte PUBREC = 5; //Publish Received (assured delivery part 1) + public static final byte PUBREL = 6; // Publish Release (assured delivery part 2) + public static final byte PUBCOMP = 7; //Publish Complete (assured delivery part 3) + public static final byte SUBSCRIBE = 8; //Client Subscribe request + public static final byte SUBACK = 9; // Subscribe Acknowledgment + public static final byte UNSUBSCRIBE = 10; //Client Unsubscribe request + public static final byte UNSUBACK = 11; // Unsubscribe Acknowledgment + public static final byte PINGREQ = 12; //PING Request + public static final byte PINGRESP = 13; //PING Response + public static final byte DISCONNECT = 14; //Client is Disconnecting + + public static enum QOSType { + + MOST_ONE, LEAST_ONE, EXACTLY_ONCE, RESERVED; + } + //type + protected boolean m_dupFlag; + protected QOSType m_qos; + protected boolean m_retainFlag; + protected int m_remainingLength; + protected byte m_messageType; + + public byte getMessageType() { + return m_messageType; + } + + public void setMessageType(byte messageType) { + this.m_messageType = messageType; + } + + public boolean isDupFlag() { + return m_dupFlag; + } + + public void setDupFlag(boolean dupFlag) { + this.m_dupFlag = dupFlag; + } + + public QOSType getQos() { + return m_qos; + } + + public void setQos(QOSType qos) { + this.m_qos = qos; + } + + public boolean isRetainFlag() { + return m_retainFlag; + } + + public void setRetainFlag(boolean retainFlag) { + this.m_retainFlag = retainFlag; + } + + /** + * TOBE used only internally + */ + public int getRemainingLength() { + return m_remainingLength; + } + + /** + * TOBE used only internally + */ + public void setRemainingLength(int remainingLength) { + this.m_remainingLength = remainingLength; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java new file mode 100644 index 00000000..9d8d4a7f --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnAckMessage.java @@ -0,0 +1,26 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * The attributes Qos, Dup and Retain aren't used. + * + * @author andrea + */ +public class ConnAckMessage extends AbstractMessage { + public static final byte CONNECTION_ACCEPTED = 0x00; + public static final byte UNNACEPTABLE_PROTOCOL_VERSION = 0x01; + public static final byte IDENTIFIER_REJECTED = 0x02; + public static final byte SERVER_UNAVAILABLE = 0x03; + public static final byte BAD_USERNAME_OR_PASSWORD = 0x04; + public static final byte NOT_AUTHORIZED = 0x05; + + private byte m_returnCode; + + public byte getReturnCode() { + return m_returnCode; + } + + public void setReturnCode(byte returnCode) { + this.m_returnCode = returnCode; + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java new file mode 100644 index 00000000..f286d682 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/ConnectMessage.java @@ -0,0 +1,140 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * The attributes Qos, Dup and Retain aren't used for Connect message + * + * @author andrea + */ +public class ConnectMessage extends AbstractMessage { + String m_protocolName; + byte m_procotolVersion; + + //Connection flags + boolean m_cleanSession; + boolean m_willFlag; + byte m_willQos; + boolean m_willRetain; + boolean m_passwordFlag; + boolean m_userFlag; + int m_keepAlive; + + //Variable part + String m_username; + String m_password; + String m_clientID; + String m_willtopic; + String m_willMessage; + + public boolean isCleanSession() { + return m_cleanSession; + } + + public void setCleanSession(boolean cleanSession) { + this.m_cleanSession = cleanSession; + } + + public int getKeepAlive() { + return m_keepAlive; + } + + public void setKeepAlive(int keepAlive) { + this.m_keepAlive = keepAlive; + } + + public boolean isPasswordFlag() { + return m_passwordFlag; + } + + public void setPasswordFlag(boolean passwordFlag) { + this.m_passwordFlag = passwordFlag; + } + + public byte getProcotolVersion() { + return m_procotolVersion; + } + + public void setProcotolVersion(byte procotolVersion) { + this.m_procotolVersion = procotolVersion; + } + + public String getProtocolName() { + return m_protocolName; + } + + public void setProtocolName(String protocolName) { + this.m_protocolName = protocolName; + } + + public boolean isUserFlag() { + return m_userFlag; + } + + public void setUserFlag(boolean userFlag) { + this.m_userFlag = userFlag; + } + + public boolean isWillFlag() { + return m_willFlag; + } + + public void setWillFlag(boolean willFlag) { + this.m_willFlag = willFlag; + } + + public byte getWillQos() { + return m_willQos; + } + + public void setWillQos(byte willQos) { + this.m_willQos = willQos; + } + + public boolean isWillRetain() { + return m_willRetain; + } + + public void setWillRetain(boolean willRetain) { + this.m_willRetain = willRetain; + } + + public String getPassword() { + return m_password; + } + + public void setPassword(String password) { + this.m_password = password; + } + + public String getUsername() { + return m_username; + } + + public void setUsername(String username) { + this.m_username = username; + } + + public String getClientID() { + return m_clientID; + } + + public void setClientID(String clientID) { + this.m_clientID = clientID; + } + + public String getWillTopic() { + return m_willtopic; + } + + public void setWillTopic(String topic) { + this.m_willtopic = topic; + } + + public String getWillMessage() { + return m_willMessage; + } + + public void setWillMessage(String willMessage) { + this.m_willMessage = willMessage; + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java new file mode 100644 index 00000000..12c4f179 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/DisconnectMessage.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * Doesn't care DUP, QOS and RETAIN flags. + * + * @author andrea + */ +public class DisconnectMessage extends ZeroLengthMessage { +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java new file mode 100644 index 00000000..eb2dae14 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/MessageIDMessage.java @@ -0,0 +1,22 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * Base class for alla the messages that carries only MessageID. (PUBACK, PUBREC, + * PUBREL, PUBCOMP, UNSUBACK) + * + * The flags dup, QOS and Retained doens't take care. + * + * @author andrea + */ +public abstract class MessageIDMessage extends AbstractMessage { + int m_messageID; + + public int getMessageID() { + return m_messageID; + } + + public void setMessageID(int messageID) { + this.m_messageID = messageID; + } + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java new file mode 100644 index 00000000..4bd1f1ba --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/PingReqMessage.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * Doesn't care DUP, QOS and RETAIN flags. + * + * @author andrea + */ +public class PingReqMessage extends ZeroLengthMessage { +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java new file mode 100644 index 00000000..aad82e57 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/PingRespMessage.java @@ -0,0 +1,8 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * Doesn't care DUP, QOS and RETAIN flags. + * @author andrea + */ +public class PingRespMessage extends ZeroLengthMessage { +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java new file mode 100644 index 00000000..5c0a87e7 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/PubAckMessage.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * Placeholder for PUBACK message. + * + * @author andrea + */ +public class PubAckMessage extends MessageIDMessage { +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java new file mode 100644 index 00000000..648e6e42 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRecMessage.java @@ -0,0 +1,10 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * Placeholder for PUBREC message. + * + * @author andrea + */ +public class PubRecMessage extends MessageIDMessage { + +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java new file mode 100644 index 00000000..aef4a5bd --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/PubRelMessage.java @@ -0,0 +1,8 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * + * @author andrea + */ +public class PubRelMessage extends MessageIDMessage { +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java new file mode 100644 index 00000000..c8b33555 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/PublishMessage.java @@ -0,0 +1,36 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * + * @author andrea + */ +public class PublishMessage extends AbstractMessage { + + private String m_topicName; + private Integer m_messageID; //could be null if Qos is == 0 + private byte[] m_payload; + + public Integer getMessageID() { + return m_messageID; + } + + public void setMessageID(Integer messageID) { + this.m_messageID = messageID; + } + + public String getTopicName() { + return m_topicName; + } + + public void setTopicName(String topicName) { + this.m_topicName = topicName; + } + + public byte[] getPayload() { + return m_payload; + } + + public void setPayload(byte[] payload) { + this.m_payload = payload; + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java new file mode 100644 index 00000000..debe6247 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/SubAckMessage.java @@ -0,0 +1,21 @@ +package org.dna.mqtt.moquette.proto.messages; + +import java.util.ArrayList; +import java.util.List; + +/** + * + * @author andrea + */ +public class SubAckMessage extends MessageIDMessage { + + List m_types = new ArrayList(); + + public List types() { + return m_types; + } + + public void addType(QOSType type) { + m_types.add(type); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java new file mode 100644 index 00000000..118a9800 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java @@ -0,0 +1,39 @@ +package org.dna.mqtt.moquette.proto.messages; + +import java.util.ArrayList; +import java.util.List; + +/** + * + * @author andrea + */ +public class SubscribeMessage extends MessageIDMessage { + + public static class Couple { + + private byte m_qos; + private String m_topic; + + public Couple(byte qos, String topic) { + m_qos = qos; + m_topic = topic; + } + + public byte getQos() { + return m_qos; + } + + public String getTopic() { + return m_topic; + } + } + private List m_subscriptions = new ArrayList(); + + public List subscriptions() { + return m_subscriptions; + } + + public void addSubscription(Couple subscription) { + m_subscriptions.add(subscription); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java new file mode 100644 index 00000000..e5f92f6d --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubAckMessage.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * + * @author andrea + */ +public class UnsubAckMessage extends MessageIDMessage { +} + diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java new file mode 100644 index 00000000..1d3912db --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/UnsubscribeMessage.java @@ -0,0 +1,20 @@ +package org.dna.mqtt.moquette.proto.messages; + +import java.util.ArrayList; +import java.util.List; + +/** + * + * @author andrea + */ +public class UnsubscribeMessage extends MessageIDMessage { + List m_types = new ArrayList(); + + public List topics() { + return m_types; + } + + public void addTopic(String type) { + m_types.add(type); + } +} diff --git a/src/main/java/org/dna/mqtt/moquette/proto/messages/ZeroLengthMessage.java b/src/main/java/org/dna/mqtt/moquette/proto/messages/ZeroLengthMessage.java new file mode 100644 index 00000000..6812dd01 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/proto/messages/ZeroLengthMessage.java @@ -0,0 +1,9 @@ +package org.dna.mqtt.moquette.proto.messages; + +/** + * + * @author andrea + */ +public abstract class ZeroLengthMessage extends AbstractMessage { + +} diff --git a/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java new file mode 100644 index 00000000..30cb6921 --- /dev/null +++ b/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -0,0 +1,54 @@ +package org.dna.mqtt.moquette.server; + +import java.util.HashMap; +import java.util.Map; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.apache.mina.core.service.IoHandlerAdapter; +import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; + +/** + * + * @author andrea + */ +public class MQTTHandler extends IoHandlerAdapter { + + Map m_clientIDs = new HashMap(); + + @Override + public void messageReceived(IoSession session, Object message) throws Exception { + AbstractMessage msg = (AbstractMessage) message; + switch (msg.getMessageType()) { + case CONNECT: + handleConnect(session, (ConnectMessage) msg); + } + } + + protected void handleConnect(IoSession session, ConnectMessage connectMessage) { + if (connectMessage.getProcotolVersion() != 0x03) { + ConnAckMessage badProto = new ConnAckMessage(); + badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); + session.write(badProto); + session.close(false); + return; + } + + //if an old client with the same ID already exists close its session.s + if (m_clientIDs.containsKey(connectMessage.getClientID())) { + m_clientIDs.get(connectMessage.getClientID()).close(false); + } + m_clientIDs.put(connectMessage.getClientID(), session); + + session.setAttribute("keepAlive", connectMessage.getKeepAlive()); + session.setAttribute("cleanSession", connectMessage.isCleanSession()); + + //TODO Handle will flag + //TODO handle user authentication + + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); + session.write(okResp); + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/ConnAckDecoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/ConnAckDecoderTest.java new file mode 100644 index 00000000..fb491acd --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/ConnAckDecoderTest.java @@ -0,0 +1,63 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.ConnAckDecoder; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.apache.mina.filter.codec.demux.MessageDecoder; +import org.apache.mina.core.buffer.IoBuffer; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +/** + * + * @author andrea + */ +public class ConnAckDecoderTest { + IoBuffer m_buff; + ConnAckDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new ConnAckDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testDecodable_OK() { + m_buff.put((byte)(AbstractMessage.CONNACK << 4)).flip(); + assertEquals(MessageDecoder.OK , m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testDecodable_NOT_OK() { + m_buff.put((byte)(AbstractMessage.CONNECT << 4)).flip(); + assertEquals(MessageDecoder.NOT_OK , m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testHeader() throws Exception { + m_buff = IoBuffer.allocate(14); + initHeader(m_buff); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_mockProtoDecoder.getMessage().getReturnCode()); + } + + private void initHeader(IoBuffer buff) { + buff.clear().put((byte)(AbstractMessage.CONNACK << 4)).put((byte)2); + //reserved + buff.put((byte)0); + //return code + buff.put(ConnAckMessage.CONNECTION_ACCEPTED); + } + +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/ConnAckEncoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/ConnAckEncoderTest.java new file mode 100644 index 00000000..b4b61cf3 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/ConnAckEncoderTest.java @@ -0,0 +1,28 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.ConnAckEncoder; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class ConnAckEncoderTest { + + @Test + public void testHeaderEncode() throws Exception { + TestUtils.MockProtocolEncoderOutput mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + ConnAckEncoder encoder = new ConnAckEncoder(); + ConnAckMessage msg = new ConnAckMessage(); + + //Exercise + encoder.encode(null, msg, mockProtoEncoder); + + //Verify + assertEquals(0x20, mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(0x02, mockProtoEncoder.getBuffer().get()); //2 byte, length + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, mockProtoEncoder.getBuffer().skip(1).get()); + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/ConnectDecoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/ConnectDecoderTest.java new file mode 100644 index 00000000..8c74f381 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/ConnectDecoderTest.java @@ -0,0 +1,188 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.ConnectDecoder; +import org.dna.mqtt.moquette.proto.Utils; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoder; + +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +public class ConnectDecoderTest { + + IoBuffer m_buff; + ConnectDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new ConnectDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testDecodable_OK() { + m_buff.put((byte)(AbstractMessage.CONNECT << 4)).flip(); + assertEquals(MessageDecoder.OK , m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testDecodable_NOT_OK() { + m_buff.put((byte)(AbstractMessage.CONNACK << 4)).flip(); + assertEquals(MessageDecoder.NOT_OK , m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testDecodable_NEED_DATA() { + m_buff = IoBuffer.allocate(0); + assertFalse(m_buff.hasRemaining()); + assertEquals(MessageDecoder.NEED_DATA , m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testDecodeRemainingLenght() { + //1 byte length + m_buff.put((byte)0x0).flip(); + assertEquals(0, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().put((byte) 0x007F).flip(); + assertEquals(127, Utils.decodeRemainingLenght(m_buff)); + + //2 byte length + m_buff.clear().put((byte) 0x80).put((byte) 0x01).flip(); + assertEquals(128, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().put((byte) 0xFF).put((byte) 0x7F).flip(); + assertEquals(16383, Utils.decodeRemainingLenght(m_buff)); + + //3 byte length + m_buff.clear().put((byte) 0x80).put((byte) 0x80).put((byte)0x01).flip(); + assertEquals(16384, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().put((byte) 0xFF).put((byte) 0xFF).put((byte)0x7F).flip(); + assertEquals(2097151, Utils.decodeRemainingLenght(m_buff)); + + //4 byte length + m_buff.clear().put((byte) 0x80).put((byte) 0x80).put((byte)0x80).put((byte)0x01).flip(); + assertEquals(2097152, Utils.decodeRemainingLenght(m_buff)); + m_buff.clear().put((byte) 0xFF).put((byte) 0xFF).put((byte) 0xFF).put((byte)0x7F).flip(); + assertEquals(268435455, Utils.decodeRemainingLenght(m_buff)); + } + + @Test + public void testBaseHeader() throws Exception { + m_buff = IoBuffer.allocate(14); + initBaseHeader(m_buff); + m_buff.flip(); + + //Excercise + m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + verifyBaseHeader(m_mockProtoDecoder.getMessage()); + } + + @Test + public void testBaseHeader_ClientID() throws UnsupportedEncodingException, Exception { + m_buff = IoBuffer.allocate(40); + initHeader(m_buff, (byte) 38); + encodeString(m_buff, "ABCDEFGH"); + encodeString(m_buff, "Topic"); + encodeString(m_buff, "Message"); + m_buff.flip(); + + //Excercise + m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + verifyBaseHeader(m_mockProtoDecoder.getMessage()); + assertEquals("ABCDEFGH", m_mockProtoDecoder.getMessage().getClientID()); + assertEquals("Topic", m_mockProtoDecoder.getMessage().getWillTopic()); + assertEquals("Message", m_mockProtoDecoder.getMessage().getWillMessage()); + } + + @Test + public void testBaseHeader_extra_with_user_pwd() throws UnsupportedEncodingException, Exception { + m_buff = IoBuffer.allocate(55); + initHeader(m_buff, (byte) 53); + encodeString(m_buff, "ABCDEFGH"); + encodeString(m_buff, "Topic"); + encodeString(m_buff, "Message"); + encodeString(m_buff, "Fakeuser"); + encodeString(m_buff, "pwd"); + m_buff.flip(); + + //Excercise + m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + verifyBaseHeader(m_mockProtoDecoder.getMessage()); + assertEquals("ABCDEFGH", m_mockProtoDecoder.getMessage().getClientID()); + assertEquals("Topic", m_mockProtoDecoder.getMessage().getWillTopic()); + assertEquals("Message", m_mockProtoDecoder.getMessage().getWillMessage()); + assertEquals("Fakeuser", m_mockProtoDecoder.getMessage().getUsername()); + assertEquals("pwd", m_mockProtoDecoder.getMessage().getPassword()); + } + + @Test + public void testBadFlagUserPwd() throws UnsupportedEncodingException, Exception { + m_buff = IoBuffer.allocate(14); + m_buff.clear().put((byte)(AbstractMessage.CONNECT << 4)).put((byte)12); + //Proto name + encodeString(m_buff, "MQIsdp"); + //version + m_buff.put((byte)3); + //conn flags + m_buff.put((byte)0x4E); //sets user to false and password to true + //keepAlive + m_buff.put((byte)0).put((byte) 0x0A); + m_buff.flip(); + + //Excercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.NOT_OK, res); + } + + /** + * Encode and insert the given stirng into the given buff + */ + private IoBuffer encodeString(IoBuffer buff, String str) throws UnsupportedEncodingException { + buff.put(Utils.encodeString(str)); + return buff; + } + + private void initBaseHeader(IoBuffer buff) throws UnsupportedEncodingException { + initHeader(buff, (byte)0x0C); + } + + private void initHeader(IoBuffer buff, byte remaingLength) throws UnsupportedEncodingException { + buff.clear().put((byte)(AbstractMessage.CONNECT << 4)).put(remaingLength); + //Proto name + encodeString(buff, "MQIsdp"); + //version + buff.put((byte)3); + //conn flags + buff.put((byte)0xCE); + //keepAlive + buff.put((byte)0).put((byte) 0x0A); + } + + 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()); + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/ConnectEncoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/ConnectEncoderTest.java new file mode 100644 index 00000000..876b6c30 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/ConnectEncoderTest.java @@ -0,0 +1,144 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.ConnectEncoder; +import org.dna.mqtt.moquette.proto.Utils; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.proto.TestUtils.*; + +/** + * + * @author andrea + */ +public class ConnectEncoderTest { + + ConnectEncoder m_encoder; + TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; + + @Before + public void setUp() { + m_encoder = new ConnectEncoder(); + m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + } + + @Test(expected=IllegalAccessException.class) + public void testEncodeRemainingLength_invalid_upper() throws IllegalAccessException { + Utils.encodeRemainingLength(Utils.MAX_LENGTH_LIMIT + 1); + } + + @Test(expected=IllegalAccessException.class) + public void testEncodeRemainingLength_invalid_lower() throws IllegalAccessException { + Utils.encodeRemainingLength(-1); + } + + @Test + public void testEncodeRemainingLenght() throws IllegalAccessException { + //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)); + } + + @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); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x10, m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(12, m_mockProtoEncoder.getBuffer().get()); //remaining length + verifyString("MQIsdp", m_mockProtoEncoder.getBuffer()); + assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //flags + assertEquals(2, m_mockProtoEncoder.getBuffer().get()); //keepAliveTimer msb + assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //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"); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x10, m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(36, m_mockProtoEncoder.getBuffer().get()); //remaining length + verifyString("MQIsdp", m_mockProtoEncoder.getBuffer()); + assertEquals(0x36, m_mockProtoEncoder.getBuffer().get()); //flags + assertEquals(2, m_mockProtoEncoder.getBuffer().get()); //keepAliveTimer msb + assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //keepAliveTimer lsb + + //Variable part + verifyString("ABCDEF", m_mockProtoEncoder.getBuffer()); + verifyString("Topic", m_mockProtoEncoder.getBuffer()); + verifyString("Message", m_mockProtoEncoder.getBuffer()); + } + + @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"); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x10, m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(48, m_mockProtoEncoder.getBuffer().get()); //remaining length + verifyString("MQIsdp", m_mockProtoEncoder.getBuffer()); + assertEquals((byte)0xF6, (byte)m_mockProtoEncoder.getBuffer().get()); //flags + assertEquals(2, m_mockProtoEncoder.getBuffer().get()); //keepAliveTimer msb + assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //keepAliveTimer lsb + + //Variable part + verifyString("ABCDEF", m_mockProtoEncoder.getBuffer()); + verifyString("Topic", m_mockProtoEncoder.getBuffer()); + verifyString("Message", m_mockProtoEncoder.getBuffer()); + verifyString("Pablo", m_mockProtoEncoder.getBuffer());//username + verifyString("PBL", m_mockProtoEncoder.getBuffer());//password + } + +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java new file mode 100644 index 00000000..43e78974 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/PubAckDecoderTest.java @@ -0,0 +1,52 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.PubAckDecoder; +import org.dna.mqtt.moquette.proto.Utils; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoder; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class PubAckDecoderTest { + IoBuffer m_buff; + PubAckDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new PubAckDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testHeader() throws Exception { + m_buff = IoBuffer.allocate(14); + int messageId = 0xAABB; + initHeader(m_buff, messageId); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals(messageId, m_mockProtoDecoder.getMessage().getMessageID()); + } + + private void initHeader(IoBuffer buff, int messageID) { + buff.clear().put((byte)(AbstractMessage.PUBACK << 4)).put((byte)2); + + //return code + Utils.writeWord(buff, messageID); + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/PubAckEncoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/PubAckEncoderTest.java new file mode 100644 index 00000000..0a690be8 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/PubAckEncoderTest.java @@ -0,0 +1,31 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.PubAckEncoder; +import org.dna.mqtt.moquette.proto.messages.PubAckMessage; +import org.junit.Test; +import static org.junit.Assert.*; +/** + * + * @author andrea + */ +public class PubAckEncoderTest { + + @Test + public void testHeaderEncode() throws Exception { + int messageID = 0xAABB; + TestUtils.MockProtocolEncoderOutput mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + PubAckEncoder encoder = new PubAckEncoder(); + PubAckMessage msg = new PubAckMessage(); + msg.setMessageID(messageID); + + //Exercise + encoder.encode(null, msg, mockProtoEncoder); + + //Verify + assertEquals(0x40, mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(0x02, mockProtoEncoder.getBuffer().get()); //2 byte, length + assertEquals((byte)0xAA, mockProtoEncoder.getBuffer().get()); + assertEquals((byte)0xBB, mockProtoEncoder.getBuffer().get()); + } + +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java new file mode 100644 index 00000000..5f8d6fb1 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/PublishDecoderTest.java @@ -0,0 +1,114 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.PublishDecoder; +import org.dna.mqtt.moquette.proto.Utils; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoder; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class PublishDecoderTest { + + IoBuffer m_buff; + PublishDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new PublishDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testDecodable_OK() { + m_buff.put((byte) (AbstractMessage.PUBLISH << 4)).flip(); + assertEquals(MessageDecoder.OK, m_msgdec.decodable(null, m_buff)); + } + + @Test + public void testHeader() throws Exception { + m_buff = IoBuffer.allocate(14); + initHeader(m_buff); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertNull(m_mockProtoDecoder.getMessage().getMessageID()); + } + + @Test + public void testHeaderWithMessageID() throws Exception { + m_buff = IoBuffer.allocate(14).setAutoExpand(true); + int messageID = 123; + initHeaderWithMessageID(m_buff, messageID); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); + } + + @Test + public void testHeaderWithMessageID_Payload() throws Exception { + m_buff = IoBuffer.allocate(14).setAutoExpand(true); + int messageID = 123; + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + initHeaderWithMessageID_Payload(m_buff, messageID, payload); + m_buff.flip(); + + //Exercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + assertNotNull(m_mockProtoDecoder.getMessage()); + assertEquals(MessageDecoder.OK, res); + assertEquals("Fake Topic", m_mockProtoDecoder.getMessage().getTopicName()); + assertEquals(messageID, (int) m_mockProtoDecoder.getMessage().getMessageID()); + TestUtils.verifyEquals(payload, m_mockProtoDecoder.getMessage().getPayload()); + } + + private void initHeader(IoBuffer buff) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")).flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4)).put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private void initHeaderWithMessageID(IoBuffer buff, int messageID) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); + Utils.writeWord(tmp, messageID); + tmp.flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 + .put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } + + private void initHeaderWithMessageID_Payload(IoBuffer buff, int messageID, byte[] payload) throws IllegalAccessException { + IoBuffer tmp = IoBuffer.allocate(4).setAutoExpand(true).put(Utils.encodeString("Fake Topic")); + Utils.writeWord(tmp, messageID); + tmp.put(payload); + tmp.flip(); + buff.clear().put((byte) (AbstractMessage.PUBLISH << 4 | 0x02)) //set Qos to 1 + .put(Utils.encodeRemainingLength(tmp.remaining())); + //topic name + buff.put(tmp); + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java new file mode 100644 index 00000000..95548b50 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/PublishEncoderTest.java @@ -0,0 +1,121 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.PublishEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.PublishMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.proto.TestUtils.*; + +/** + * + * @author andrea + */ +public class PublishEncoderTest { + + PublishEncoder m_encoder; + TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; + + @Before + public void setUp() { + m_encoder = new PublishEncoder(); + m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + } + + @Test + public void testEncodeWithQos_0_noMessageID() throws Exception { + String topic = "pictures/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.MOST_ONE); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x30, m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(24, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //Variable part + verifyString(topic, m_mockProtoEncoder.getBuffer()); + verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); + } + + @Test + public void testEncodeWithQos_1_MessageID() throws Exception { + String topic = "pictures/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(topic); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals(0x32, m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(26, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //Variable part + verifyString(topic, m_mockProtoEncoder.getBuffer()); + assertEquals(0, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB + assertEquals(1, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB + verifyBuff(payload.length, payload, m_mockProtoEncoder.getBuffer()); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncodeWithQos_1_noMessageID_fail() throws Exception { + String topic = "pictures/photos"; + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setTopicName(topic); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_null_topic() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(1); + msg.setTopicName(null); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_qos_reserved() throws Exception { + PublishMessage msg = new PublishMessage(); + msg.setQos(QOSType.RESERVED); + msg.setMessageID(1); + msg.setTopicName(null); + + //variable part + byte[] payload = new byte[]{0x0A, 0x0B, 0x0C}; + msg.setPayload(payload); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java new file mode 100644 index 00000000..0b376699 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/SubAckDecoderTest.java @@ -0,0 +1,60 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.SubAckDecoder; +import org.dna.mqtt.moquette.proto.Utils; +import java.util.List; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class SubAckDecoderTest { + + IoBuffer m_buff; + SubAckDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(7); + m_msgdec = new SubAckDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testBadQos() throws Exception { + initHeaderQos(m_buff, 0xAABB, QOSType.LEAST_ONE, QOSType.MOST_ONE, QOSType.MOST_ONE); + m_buff.flip(); + + //Excercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + assertEquals(MessageDecoderResult.OK, res); + assertEquals(0xAABB, m_mockProtoDecoder.getMessage().getMessageID()); + List qoses = m_mockProtoDecoder.getMessage().types(); + assertEquals(3, qoses.size()); + assertEquals(QOSType.LEAST_ONE, qoses.get(0)); + assertEquals(QOSType.MOST_ONE, qoses.get(1)); + assertEquals(QOSType.MOST_ONE, qoses.get(2)); + } + + private void initHeaderQos(IoBuffer buff, int messageID, QOSType... qoss) throws IllegalAccessException { + buff.clear().put((byte) (AbstractMessage.SUBSCRIBE << 4)). + put(Utils.encodeRemainingLength(3)); + + Utils.writeWord(buff, messageID); + for (QOSType qos : qoss) { + buff.put((byte)qos.ordinal()); + } + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/SubAckEncoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/SubAckEncoderTest.java new file mode 100644 index 00000000..5d681608 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/SubAckEncoderTest.java @@ -0,0 +1,58 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.SubAckEncoder; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class SubAckEncoderTest { + SubAckEncoder m_encoder; + TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; + + @Before + public void setUp() { + m_encoder = new SubAckEncoder(); + m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncodeWithNoQoss() throws Exception { + SubAckMessage msg = new SubAckMessage(); + msg.setMessageID(123); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test + public void testEncodeWithMultipleQos() throws Exception { + SubAckMessage msg = new SubAckMessage(); + + int messageID = 0xAABB; + msg.setMessageID(messageID); + msg.addType(QOSType.MOST_ONE); + msg.addType(QOSType.LEAST_ONE); + msg.addType(QOSType.EXACTLY_ONCE); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals((byte) (AbstractMessage.SUBACK << 4 ), m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(5, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //Variable part + assertEquals((byte)0xAA, m_mockProtoEncoder.getBuffer().get()); //MessageID MSB + assertEquals((byte)0xBB, m_mockProtoEncoder.getBuffer().get()); //MessageID LSB + assertEquals((byte)QOSType.MOST_ONE.ordinal(), m_mockProtoEncoder.getBuffer().get()); + assertEquals((byte)QOSType.LEAST_ONE.ordinal(), m_mockProtoEncoder.getBuffer().get()); + assertEquals((byte)QOSType.EXACTLY_ONCE.ordinal(), m_mockProtoEncoder.getBuffer().get()); + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/SubscribeDecoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/SubscribeDecoderTest.java new file mode 100644 index 00000000..ebae75f3 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/SubscribeDecoderTest.java @@ -0,0 +1,79 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.SubscribeDecoder; +import org.dna.mqtt.moquette.proto.Utils; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage.Couple; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class SubscribeDecoderTest { + + IoBuffer m_buff; + SubscribeDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new SubscribeDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testBadQos() throws Exception { + m_buff = IoBuffer.allocate(2); + initHeaderBadQos(m_buff); + m_buff.flip(); + + //Excercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + assertEquals(MessageDecoderResult.NOT_OK, res); + } + + @Test + public void testMultiTopic() throws Exception { + m_buff = IoBuffer.allocate(4).setAutoExpand(true); + Couple c1 = new Couple((byte)2, "a/b"); + Couple c2 = new Couple((byte)1, "c/d/e"); + initMultiTopic(m_buff, 123, c1, c2); + m_buff.flip(); + + //Excercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + assertEquals(MessageDecoderResult.OK, res); + assertEquals(2, m_mockProtoDecoder.getMessage().subscriptions().size()); + } + + private void initHeaderBadQos(IoBuffer buff) { + buff.clear().put((byte)(AbstractMessage.SUBSCRIBE << 4)).put((byte)0); + } + + private void initMultiTopic(IoBuffer buff, int messageID, Couple... topics) throws IllegalAccessException { + IoBuffer topicBuffer = IoBuffer.allocate(4).setAutoExpand(true); + Utils.writeWord(topicBuffer, messageID); + for (Couple couple : topics) { + topicBuffer.put(Utils.encodeString(couple.getTopic())); + topicBuffer.put((byte)couple.getQos()); + } + topicBuffer.flip(); + + buff.clear().put((byte)(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02)) + .put(Utils.encodeRemainingLength(topicBuffer.remaining())); + buff.put(topicBuffer); + } +} + diff --git a/src/test/java/org/dna/mqtt/moquette/proto/SubscribeEncoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/SubscribeEncoderTest.java new file mode 100644 index 00000000..3a4cc713 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/SubscribeEncoderTest.java @@ -0,0 +1,70 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.UnsubscribeEncoder; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.proto.TestUtils.*; + +/** + * + * @author andrea + */ +public class SubscribeEncoderTest { + UnsubscribeMessage m_msg; + UnsubscribeEncoder m_encoder; + TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; + + @Before + public void setUp() { + m_encoder = new UnsubscribeEncoder(); + m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + m_msg = new UnsubscribeMessage(); + m_msg.setMessageID(0xAABB); + } + + @Test + public void testEncodeWithMultiTopic() throws Exception { + m_msg.setQos(QOSType.LEAST_ONE); + + //variable part + String topic1 = "a/b"; + String topic2 = "a/b/c"; + m_msg.addTopic(topic1); + m_msg.addTopic(topic2); + + //Exercise + m_encoder.encode(null, m_msg, m_mockProtoEncoder); + + //Verify + assertEquals((byte)0xA2, (byte)m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(18, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //verify M1ssageID + assertEquals((byte)0xAA, m_mockProtoEncoder.getBuffer().get()); + assertEquals((byte)0xBB, m_mockProtoEncoder.getBuffer().get()); + + //Variable part + verifyString(topic1, m_mockProtoEncoder.getBuffer()); + verifyString(topic2, m_mockProtoEncoder.getBuffer()); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_topics() throws Exception { + m_msg.setQos(QOSType.LEAST_ONE); + + //Exercise + m_encoder.encode(null, m_msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_badQos() throws Exception { + m_msg.setQos(QOSType.EXACTLY_ONCE); + + //Exercise + m_encoder.encode(null, m_msg, m_mockProtoEncoder); + } + +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java b/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java new file mode 100644 index 00000000..16ce1c73 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/TestUtils.java @@ -0,0 +1,98 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.Utils; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import java.io.UnsupportedEncodingException; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.core.filterchain.IoFilter.NextFilter; +import org.apache.mina.core.future.WriteFuture; +import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.codec.ProtocolDecoderOutput; +import org.apache.mina.filter.codec.ProtocolEncoderOutput; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class TestUtils { + + static final class MockProtocolDecoderOutput implements ProtocolDecoderOutput { + + protected T m_connMessage; + + public T getMessage() { + return m_connMessage; + } + + public void write(Object message) { + m_connMessage = (T) message; + } + + public void flush(NextFilter nextFilter, IoSession session) { + throw new UnsupportedOperationException("Not supported yet."); + } + } + + + static final class MockProtocolEncoderOutput implements ProtocolEncoderOutput { + + protected IoBuffer m_buffer; + + public IoBuffer getBuffer() { + return m_buffer; + } + + public void write(Object buffer) { + m_buffer = (IoBuffer) buffer; + } + + public void mergeAll() { + throw new UnsupportedOperationException("Not supported yet."); + } + + public WriteFuture flush() { + throw new UnsupportedOperationException("Not supported yet."); + } + } + + /** + * Verify the presence of the given string starting from the current position + * inside the buffer. + */ + static void verifyString(String str, IoBuffer buff) throws UnsupportedEncodingException { + IoBuffer tmpBuff = IoBuffer.allocate(2).setAutoExpand(true); + byte[] raw = str.getBytes("UTF-8"); + Utils.writeWord(tmpBuff, raw.length); + tmpBuff.put(raw).flip(); + + verifyIoBuffer(tmpBuff, buff); + } + + static void verifyIoBuffer(IoBuffer expected, IoBuffer found) { + while(expected.hasRemaining()) { + assertEquals(expected.get(), found.get()); + } + } + + /** + * Verify that the given bytes buffer og the given numBytes length is present + * in the buff starting from the current position. + */ + static void verifyBuff(int numBytes, byte[] bytes, IoBuffer buff) { + assertTrue(numBytes <= buff.remaining()); + byte[] toCheck = new byte[numBytes]; + buff.get(toCheck); + + for (int i = 0; i < numBytes; i++) { + assertEquals(bytes[i], toCheck[i]); + } + } + + static void verifyEquals(byte[] expected, byte[] found) { + assertEquals(expected.length, found.length); + for (int i = 0; i < expected.length; i++) { + assertEquals(expected[i], found[i]); + } + } +} diff --git a/src/test/java/org/dna/mqtt/moquette/proto/UnsubscribeDecoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/UnsubscribeDecoderTest.java new file mode 100644 index 00000000..ee402018 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/UnsubscribeDecoderTest.java @@ -0,0 +1,79 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.Utils; +import org.dna.mqtt.moquette.proto.UnsubscribeDecoder; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; +import org.apache.mina.core.buffer.IoBuffer; +import org.apache.mina.filter.codec.demux.MessageDecoderResult; +import org.dna.mqtt.moquette.proto.TestUtils.MockProtocolDecoderOutput; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class UnsubscribeDecoderTest { + + IoBuffer m_buff; + UnsubscribeDecoder m_msgdec; + MockProtocolDecoderOutput m_mockProtoDecoder; + + @Before + public void setUp() { + m_buff = IoBuffer.allocate(4); + m_msgdec = new UnsubscribeDecoder(); + m_mockProtoDecoder = new MockProtocolDecoderOutput(); + } + + @Test + public void testBadQos() throws Exception { + m_buff = IoBuffer.allocate(2); + initHeaderBadQos(m_buff); + m_buff.flip(); + + //Excercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + assertEquals(MessageDecoderResult.NOT_OK, res); + } + + @Test + public void testMultiTopic() throws Exception { + m_buff = IoBuffer.allocate(4).setAutoExpand(true); + String topic1 = "a/b"; + String topic2 = "c/d/e"; + initMultiTopic(m_buff, 123, topic1, topic2); + m_buff.flip(); + + //Excercise + MessageDecoderResult res = m_msgdec.decode(null, m_buff, m_mockProtoDecoder); + + //Verify + assertEquals(MessageDecoderResult.OK, res); + assertEquals(2, m_mockProtoDecoder.getMessage().topics().size()); + assertEquals(topic1, m_mockProtoDecoder.getMessage().topics().get(0)); + assertEquals(topic2, m_mockProtoDecoder.getMessage().topics().get(1)); + } + + private void initHeaderBadQos(IoBuffer buff) { + buff.clear().put((byte)(AbstractMessage.UNSUBSCRIBE << 4)).put((byte)0); + } + + private void initMultiTopic(IoBuffer buff, int messageID, String... topics) throws IllegalAccessException { + IoBuffer topicBuffer = IoBuffer.allocate(4).setAutoExpand(true); + Utils.writeWord(topicBuffer, messageID); + for (String topic : topics) { + topicBuffer.put(Utils.encodeString(topic)); + } + topicBuffer.flip(); + + buff.clear().put((byte)(AbstractMessage.SUBSCRIBE << 4 | (byte)0x02)) + .put(Utils.encodeRemainingLength(topicBuffer.remaining())); + buff.put(topicBuffer); + } +} + diff --git a/src/test/java/org/dna/mqtt/moquette/proto/UnsubscribeEncoderTest.java b/src/test/java/org/dna/mqtt/moquette/proto/UnsubscribeEncoderTest.java new file mode 100644 index 00000000..dc37906b --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/proto/UnsubscribeEncoderTest.java @@ -0,0 +1,77 @@ +package org.dna.mqtt.moquette.proto; + +import org.dna.mqtt.moquette.proto.SubscribeEncoder; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage.Couple; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.junit.Before; +import org.junit.Test; +import static org.junit.Assert.*; +import static org.dna.mqtt.moquette.proto.TestUtils.*; + +/** + * + * @author andrea + */ +public class UnsubscribeEncoderTest { + + SubscribeEncoder m_encoder; + TestUtils.MockProtocolEncoderOutput m_mockProtoEncoder; + + @Before + public void setUp() { + m_encoder = new SubscribeEncoder(); + m_mockProtoEncoder = new TestUtils.MockProtocolEncoderOutput(); + } + + @Test + public void testEncodeWithMultiTopic() throws Exception { + SubscribeMessage msg = new SubscribeMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(0xAABB); + + //variable part + Couple c1 = new Couple((byte)1, "a/b"); + Couple c2 = new Couple((byte)0, "a/b/c"); + msg.addSubscription(c1); + msg.addSubscription(c2); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + + //Verify + assertEquals((byte)0x82, (byte)m_mockProtoEncoder.getBuffer().get()); //1 byte + assertEquals(20, m_mockProtoEncoder.getBuffer().get()); //remaining length + + //verify M1ssageID + assertEquals((byte)0xAA, m_mockProtoEncoder.getBuffer().get()); + assertEquals((byte)0xBB, m_mockProtoEncoder.getBuffer().get()); + + //Variable part + verifyString(c1.getTopic(), m_mockProtoEncoder.getBuffer()); + assertEquals(c1.getQos(), m_mockProtoEncoder.getBuffer().get()); + verifyString(c2.getTopic(), m_mockProtoEncoder.getBuffer()); + assertEquals(c2.getQos(), m_mockProtoEncoder.getBuffer().get()); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_empty_subscription() throws Exception { + SubscribeMessage msg = new SubscribeMessage(); + msg.setQos(QOSType.LEAST_ONE); + msg.setMessageID(0xAABB); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + + @Test(expected = IllegalArgumentException.class) + public void testEncode_badQos() throws Exception { + SubscribeMessage msg = new SubscribeMessage(); + msg.setQos(QOSType.EXACTLY_ONCE); + msg.setMessageID(0xAABB); + + //Exercise + m_encoder.encode(null, msg, m_mockProtoEncoder); + } + +} diff --git a/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java new file mode 100644 index 00000000..73a96998 --- /dev/null +++ b/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -0,0 +1,42 @@ +package org.dna.mqtt.moquette.server; + +import org.apache.mina.core.write.WriteRequest; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.apache.mina.core.filterchain.IoFilterAdapter; +import org.apache.mina.core.session.DummySession; +import org.apache.mina.core.session.IoSession; +import org.dna.mqtt.moquette.proto.messages.ConnectMessage; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class MQTTHandlerTest { + + @Test + public void testHandleConnect_BadProtocol() { + MQTTHandler h = new MQTTHandler(); + ConnectMessage connMsg = new ConnectMessage(); + connMsg.setProcotolVersion((byte) 0x02); + + IoSession session = new DummySession(); + session.getFilterChain().addFirst("MessageCatcher", new IoFilterAdapter() { + + @Override + public void filterWrite(NextFilter nextFilter, IoSession session, + WriteRequest writeRequest) throws Exception { + try { + ConnAckMessage buf = (ConnAckMessage) writeRequest.getMessage(); + assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, buf.getReturnCode()); + } catch (Exception ex) { + throw new AssertionError("Wrong return code"); + } + } + }); + + //Exercise + h.handleConnect(session, connMsg); + } +} \ No newline at end of file