Changed lib structure to be modular

2012-05-10 22:00:57 +02:00
parent 0dc403d278
commit 2cbf237887
75 changed files with 87 additions and 65 deletions
@@ -0,0 +1,75 @@
+package org.dna.mqtt.moquette.client;
+
+import java.net.InetSocketAddress;
+import java.util.logging.Logger;
+import org.apache.mina.core.RuntimeIoException;
+import org.apache.mina.core.future.ConnectFuture;
+import org.apache.mina.core.session.IoSession;
+import org.apache.mina.filter.codec.ProtocolCodecFilter;
+import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
+import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
+import org.apache.mina.filter.logging.LoggingFilter;
+import org.apache.mina.transport.socket.nio.NioSocketConnector;
+import org.dna.mqtt.moquette.proto.ConnAckDecoder;
+import org.dna.mqtt.moquette.proto.ConnectDecoder;
+import org.dna.mqtt.moquette.proto.ConnectEncoder;
+import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
+import org.dna.mqtt.moquette.server.Server;
+
+/**
+ * Simple client to connect to a server
+ * 
+ * 
+ * @author andrea
+ */
+public class Client {
+    
+    private static final Logger LOG = Logger.getLogger(Client.class.getName());
+
+    private static final String HOSTNAME = /*"localhost"*/ "127.0.0.1";
+    private static final int PORT = Server.PORT;
+    private static final long CONNECT_TIMEOUT = 30 * 1000L; // 30 seconds
+
+    public static void main(String[] args) throws Throwable {
+        DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
+        decoder.addMessageDecoder(new ConnAckDecoder());
+        
+        DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
+        encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
+        
+        
+        NioSocketConnector connector = new NioSocketConnector();
+
+        // Configure the service.
+        connector.setConnectTimeoutMillis(CONNECT_TIMEOUT);
+        connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder));
+        connector.getFilterChain().addLast("logger", new LoggingFilter());
+        connector.setHandler(new  ClientMQTTHandler());
+
+        IoSession session;
+        for (;;) {
+            try {
+                ConnectFuture future = connector.connect(new InetSocketAddress(HOSTNAME, PORT));
+                LOG.info("Client waiting to connect to server");
+                future.awaitUninterruptibly();
+                session = future.getSession();
+                break;
+            } catch (RuntimeIoException e) {
+                System.err.println("Failed to connect.");
+                e.printStackTrace();
+                Thread.sleep(5000);
+            }
+        }
+        
+        //send a message over the session
+        ConnectMessage connMsg = new ConnectMessage();
+        connMsg.setKeepAlive(3);
+        session.write(connMsg);
+        LOG.info("Client wrote message to server");
+
+        // wait until the summation is done
+        session.getCloseFuture().awaitUninterruptibly();
+        connector.dispose();
+    }
+    
+}
@@ -0,0 +1,12 @@
+package org.dna.mqtt.moquette.client;
+
+import org.apache.mina.core.service.IoHandlerAdapter;
+
+/**
+ *
+ * @author andrea
+ */
+public class ClientMQTTHandler extends IoHandlerAdapter {
+//    public void sessionOpened(IoSession session) {}
+//    public void messageReceived(IoSession session, Object message) {
+}
@@ -0,0 +1,20 @@
+package org.dna.mqtt.moquette.messaging.spi;
+
+/**
+ * Interface to the underling messaging system used to publish, subscribe.
+ * 
+ * @author andrea
+ */
+public class IMessaging {
+
+    /**
+     * Publish a message on the messaging system. The qos is the level of quality 
+     * the retain flag is true iff the message should be persisted for future 
+     * notification also after all the current subscriber to the given topic has
+     * been notified.
+     */
+    public void publish(String topic, Object message, byte qos, boolean retain) {
+        throw new UnsupportedOperationException("Not yet implemented");
+    }
+    
+}
@@ -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,72 @@
+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"));
+        
+        //version 
+        staticHeaderBuff.put((byte)0x03);
+        
+        //connection flags and Strings
+        byte connectionFlags = 0;
+        if (message.isCleanSession()) {
+            connectionFlags |= 0x02;
+        }
+        if (message.isWillFlag()) {
+            connectionFlags |= 0x04;
+        }
+        connectionFlags |= ((message.getWillQos() & 0x03) << 3);
+        if (message.isWillRetain()) {
+            connectionFlags |= 0x020;
+        }
+        if (message.isPasswordFlag()) {
+            connectionFlags |= 0x040;
+        }
+        if (message.isUserFlag()) {
+            connectionFlags |= 0x080;
+        }
+        staticHeaderBuff.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,11 @@
+package org.dna.mqtt.moquette.server;
+
+/**
+ * username and password checker
+ * 
+ * @author andrea
+ */
+public interface IAuthenticator {
+
+    boolean checkValid(String username, String password);
+}
@@ -0,0 +1,102 @@
+package org.dna.mqtt.moquette.server;
+
+import org.dna.mqtt.moquette.messaging.spi.IMessaging;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Logger;
+import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
+import org.apache.mina.core.service.IoHandlerAdapter;
+import org.apache.mina.core.session.IdleStatus;
+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 {
+
+    private static final Logger LOG = Logger.getLogger(MQTTHandler.class.getName());
+    Map<String, IoSession> m_clientIDs = new HashMap<String, IoSession>();
+    private IMessaging m_messaging;
+    private IAuthenticator m_authenticator;
+
+    @Override
+    public void messageReceived(IoSession session, Object message) throws Exception {
+        AbstractMessage msg = (AbstractMessage) message;
+        LOG.fine("Received a message of type " + msg.getMessageType());
+        switch (msg.getMessageType()) {
+            case CONNECT:
+                handleConnect(session, (ConnectMessage) msg);
+        }
+    }
+
+    protected void handleConnect(IoSession session, ConnectMessage msg) {
+        if (msg.getProcotolVersion() != 0x03) {
+            ConnAckMessage badProto = new ConnAckMessage();
+            badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
+            session.write(badProto);
+            session.close(false);
+            return;
+        }
+
+        if (msg.getClientID().length() > 23) {
+            ConnAckMessage okResp = new ConnAckMessage();
+            okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
+            session.write(okResp);
+            return;
+        }
+
+        //if an old client with the same ID already exists close its session.
+        if (m_clientIDs.containsKey(msg.getClientID())) {
+            m_clientIDs.get(msg.getClientID()).close(false);
+        }
+        m_clientIDs.put(msg.getClientID(), session);
+
+        int keepAlive = msg.getKeepAlive();
+        session.setAttribute("keepAlive", keepAlive);
+        session.setAttribute("cleanSession", msg.isCleanSession());
+
+        session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));
+
+        //Handle will flag
+        if (msg.isWillFlag()) {
+            m_messaging.publish(msg.getWillTopic(), msg.getWillMessage(), msg.getWillQos(), msg.isWillRetain());
+        }
+
+        //handle user authentication
+        if (msg.isUserFlag()) {
+            String pwd = null;
+            if (msg.isPasswordFlag()) {
+                pwd = msg.getPassword();
+            }
+            if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
+                ConnAckMessage okResp = new ConnAckMessage();
+                okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
+                session.write(okResp);
+                return;
+            }
+        }
+
+        //TODO handle clean session flag
+
+        ConnAckMessage okResp = new ConnAckMessage();
+        okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
+        session.write(okResp);
+    }
+
+    @Override
+    public void sessionIdle(IoSession session, IdleStatus status) {
+        session.close(false);
+    }
+
+    public void setMessaging(IMessaging messaging) {
+        m_messaging = messaging;
+    }
+
+    public void setAuthenticator(IAuthenticator authenticator) {
+        m_authenticator = authenticator;
+    }
+}
@@ -0,0 +1,47 @@
+package org.dna.mqtt.moquette.server;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.logging.Logger;
+import org.apache.mina.core.service.IoAcceptor;
+import org.apache.mina.core.session.IdleStatus;
+import org.apache.mina.filter.codec.ProtocolCodecFilter;
+import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder;
+import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder;
+import org.apache.mina.filter.logging.LoggingFilter;
+import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
+import org.dna.mqtt.moquette.proto.ConnAckEncoder;
+import org.dna.mqtt.moquette.proto.ConnectDecoder;
+import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
+/**
+ * Launch a  configured version of the server.
+ * @author andrea
+ */
+public class Server {
+    
+    private static final Logger LOG = Logger.getLogger(Server.class.getName());
+    
+    public static final int PORT = 9191;
+    public static final int DEFAULT_CONNECT_TIMEOUT = 10;
+    
+    public static void main(String[] args) throws IOException {
+        DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder();
+        decoder.addMessageDecoder(new ConnectDecoder());
+        
+        DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
+//        encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
+        encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());
+        
+        IoAcceptor acceptor = new NioSocketAcceptor();
+
+        acceptor.getFilterChain().addLast( "logger", new LoggingFilter() );
+        acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(encoder, decoder));
+
+        acceptor.setHandler( new MQTTHandler() );
+	acceptor.getSessionConfig().setReadBufferSize( 2048 );
+        acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, DEFAULT_CONNECT_TIMEOUT );
+        acceptor.bind( new InetSocketAddress(PORT) );
+        LOG.info("Server binded");
+    }
+    
+}