First import of moquette proto parser
This commit is contained in:
selva.andre
Binary file not shown.
@@ -0,0 +1,56 @@
|
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<groupId>org.dna.mqtt</groupId>
|
||||
<artifactId>moquette</artifactId>
|
||||
<version>0.1-SNAPSHOT</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<name>moquette</name>
|
||||
<url>http://maven.apache.org</url>
|
||||
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.mina</groupId>
|
||||
<artifactId>mina-core</artifactId>
|
||||
<version>2.0.4</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<version>4.8.2</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<source>1.5</source>
|
||||
<target>1.5</target>
|
||||
</configuration>
|
||||
</plugin>
|
||||
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<configuration>
|
||||
<excludes>
|
||||
<exclude>org/dna/mqtt/moquette/proto/TestUtils.java</exclude>
|
||||
<exclude>org/dna/mqtt/moquette/proto/TestUtils$*</exclude>
|
||||
</excludes>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<ConnAckMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<ConnectMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<DisconnectMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<PingReqMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<PingRespMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<PubAckMessage> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<PubCompMessage> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<PubRecMessage> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<PubRelMessage> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<PublishMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<SubAckMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<SubscribeMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<UnsubAckMessage> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<UnsubscribeMessage> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 {
|
||||
}
|
||||
@@ -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 {
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package org.dna.mqtt.moquette.proto.messages;
|
||||
|
||||
/**
|
||||
* Placeholder for PUBACK message.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PubAckMessage extends MessageIDMessage {
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package org.dna.mqtt.moquette.proto.messages;
|
||||
|
||||
/**
|
||||
* Placeholder for PUBREC message.
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PubRecMessage extends MessageIDMessage {
|
||||
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package org.dna.mqtt.moquette.proto.messages;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class PubRelMessage extends MessageIDMessage {
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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<QOSType> m_types = new ArrayList<QOSType>();
|
||||
|
||||
public List<QOSType> types() {
|
||||
return m_types;
|
||||
}
|
||||
|
||||
public void addType(QOSType type) {
|
||||
m_types.add(type);
|
||||
}
|
||||
}
|
||||
@@ -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<Couple> m_subscriptions = new ArrayList<Couple>();
|
||||
|
||||
public List<Couple> subscriptions() {
|
||||
return m_subscriptions;
|
||||
}
|
||||
|
||||
public void addSubscription(Couple subscription) {
|
||||
m_subscriptions.add(subscription);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package org.dna.mqtt.moquette.proto.messages;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public class UnsubAckMessage extends MessageIDMessage {
|
||||
}
|
||||
|
||||
@@ -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<String> m_types = new ArrayList<String>();
|
||||
|
||||
public List<String> topics() {
|
||||
return m_types;
|
||||
}
|
||||
|
||||
public void addTopic(String type) {
|
||||
m_types.add(type);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package org.dna.mqtt.moquette.proto.messages;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author andrea
|
||||
*/
|
||||
public abstract class ZeroLengthMessage extends AbstractMessage {
|
||||
|
||||
}
|
||||
@@ -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<String, IoSession> m_clientIDs = new HashMap<String, IoSession>();
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
@@ -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<ConnAckMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new ConnAckDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<ConnAckMessage>();
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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<ConnectMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new ConnectDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<ConnectMessage>();
|
||||
}
|
||||
|
||||
@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());
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<PubAckMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new PubAckDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<PubAckMessage>();
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<PublishMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new PublishDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<PublishMessage>();
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<SubAckMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(7);
|
||||
m_msgdec = new SubAckDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<SubAckMessage>();
|
||||
}
|
||||
|
||||
@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<QOSType> 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());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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<SubscribeMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new SubscribeDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<SubscribeMessage>();
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<T extends AbstractMessage> 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]);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<UnsubscribeMessage> m_mockProtoDecoder;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
m_buff = IoBuffer.allocate(4);
|
||||
m_msgdec = new UnsubscribeDecoder();
|
||||
m_mockProtoDecoder = new MockProtocolDecoderOutput<UnsubscribeMessage>();
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user