Rearchitected the broker, switching from ProtocolProcessor and publisher to MQTTConnection, Session and PostOffice

broker/src/main/java/io/moquette/broker/Authorizator.java

@@ -0,0 +1,80 @@
+package io.moquette.broker;
+
+import io.moquette.spi.ClientSession;
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.security.IAuthorizatorPolicy;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
+import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static io.moquette.spi.impl.Utils.messageId;
+import static io.netty.handler.codec.mqtt.MqttQoS.FAILURE;
+
+final class Authorizator {
+
+    private static final Logger LOG = LoggerFactory.getLogger(Authorizator.class);
+
+    private final IAuthorizatorPolicy policy;
+
+    Authorizator(IAuthorizatorPolicy policy) {
+        this.policy = policy;
+    }
+
+    /**
+     * @param clientID
+     *            the clientID
+     * @param username
+     *            the username
+     * @param msg
+     *            the subscribe message to verify
+     * @return the list of verified topics for the given subscribe message.
+     */
+    List<MqttTopicSubscription> verifyTopicsReadAccess(String clientID, String username, MqttSubscribeMessage msg) {
+        List<MqttTopicSubscription> ackTopics = new ArrayList<>();
+
+        final int messageId = messageId(msg);
+        for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) {
+            Topic topic = new Topic(req.topicName());
+            if (!policy.canRead(topic, username, clientID)) {
+                // send SUBACK with 0x80, the user hasn't credentials to read the topic
+                LOG.warn("Client does not have read permissions on the topic CId={}, username: {}, messageId: {}, " +
+                         "topic: {}", clientID, username, messageId, topic);
+                ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE));
+            } else {
+                MqttQoS qos;
+                if (topic.isValid()) {
+                    LOG.debug("Client will be subscribed to the topic CId={}, username: {}, messageId: {}, topic: {}",
+                              clientID, username, messageId, topic);
+                    qos = req.qualityOfService();
+                } else {
+                    LOG.warn("Topic filter is not valid CId={}, username: {}, messageId: {}, topic: {}", clientID,
+                             username, messageId, topic);
+                    qos = FAILURE;
+                }
+                ackTopics.add(new MqttTopicSubscription(topic.toString(), qos));
+            }
+        }
+        return ackTopics;
+    }
+
+    /**
+     * Ask the authorization policy if the topic can be used in a publish.
+     *
+     * @param topic
+     *            the topic to write to.
+     * @param user
+     *            the user
+     * @param client
+     *            the client
+     * @return true if the user from client can publish data on topic.
+     */
+    boolean canWrite(Topic topic, String user, String client) {
+        return policy.canWrite(topic, user, client);
+    }
+
+}

broker/src/main/java/io/moquette/broker/BrokerConfiguration.java

@@ -0,0 +1,36 @@
+package io.moquette.broker;
+
+import io.moquette.BrokerConstants;
+import io.moquette.server.config.IConfig;
+
+class BrokerConfiguration {
+
+    private final boolean allowAnonymous;
+    private final boolean allowZeroByteClientId;
+    private final boolean reauthorizeSubscriptionsOnConnect;
+
+    BrokerConfiguration(IConfig props) {
+        allowAnonymous = props.boolProp(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
+        allowZeroByteClientId = props.boolProp(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, false);
+        reauthorizeSubscriptionsOnConnect = props.boolProp(BrokerConstants.REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT, false);
+    }
+
+    public BrokerConfiguration(boolean allowAnonymous, boolean allowZeroByteClientId,
+                               boolean reauthorizeSubscriptionsOnConnect) {
+        this.allowAnonymous = allowAnonymous;
+        this.allowZeroByteClientId = allowZeroByteClientId;
+        this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect;
+    }
+
+    public boolean isAllowAnonymous() {
+        return allowAnonymous;
+    }
+
+    public boolean isAllowZeroByteClientId() {
+        return allowZeroByteClientId;
+    }
+
+    public boolean isReauthorizeSubscriptionsOnConnect() {
+        return reauthorizeSubscriptionsOnConnect;
+    }
+}

broker/src/main/java/io/moquette/broker/IRetainedRepository.java

@@ -0,0 +1,14 @@
+package io.moquette.broker;
+
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+
+public interface IRetainedRepository {
+    void cleanRetained(Topic topic);
+
+    void retain(Topic topic, MqttPublishMessage msg);
+
+    boolean isEmtpy();
+
+    MqttPublishMessage retainedOnTopic(String topic);
+}

broker/src/main/java/io/moquette/broker/InflightResender.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright (c) 2012-2016 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.moquette.broker;
+
+import io.moquette.server.netty.NettyUtils;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.concurrent.EventExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Resend inflight not ack'ed publish packets (QoS1 PUB and QoS2 PUB/PUBREL). It's inspired by IdleStateHandler but it's
+ * specialized version, just invoking Session's resendInflightNotAcked by the channel after a period.
+ */
+public class InflightResender extends ChannelDuplexHandler {
+
+    /**
+     * Placeholder event to resend not-acked publish messages in the in flight window.
+     * */
+    public static class ResendNotAckedPublishes {
+    }
+
+    private final class WriterIdleTimeoutTask implements Runnable {
+
+        private final ChannelHandlerContext ctx;
+
+        WriterIdleTimeoutTask(ChannelHandlerContext ctx) {
+            this.ctx = ctx;
+        }
+
+        @Override
+        public void run() {
+            if (!ctx.channel().isOpen()) {
+                return;
+            }
+            long nextDelay = resenderTimeNanos - (System.nanoTime() - lastExecutionTime);
+            if (nextDelay <= 0) {
+                // Writer is idle - set a new timeout and notify the callback.
+                resenderTimeout = ctx.executor().schedule(this, resenderTimeNanos, TimeUnit.NANOSECONDS);
+                try {
+                    resendNotAcked(ctx/* , event */);
+                } catch (Throwable t) {
+                    ctx.fireExceptionCaught(t);
+                }
+            } else {
+                // Write occurred before the timeout - set a new timeout with shorter delay.
+                resenderTimeout = ctx.executor().schedule(this, nextDelay, TimeUnit.NANOSECONDS);
+            }
+        }
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(InflightResender.class);
+    private static final long MIN_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
+
+    private final long resenderTimeNanos;
+    volatile ScheduledFuture<?> resenderTimeout;
+    volatile long lastExecutionTime;
+
+    private volatile int state; // 0 - none, 1 - initialized, 2 - destroyed
+
+    public InflightResender(long writerIdleTime, TimeUnit unit) {
+        if (unit == null) {
+            throw new NullPointerException("unit");
+        }
+        resenderTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
+    }
+
+    @Override
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
+        if (ctx.channel().isActive() && ctx.channel().isRegistered()) {
+            // channelActive() event has been fired already, which means this.channelActive() will
+            // not be invoked. We have to initialize here instead.
+            initialize(ctx);
+        } else {
+            // channelActive() event has not been fired yet. this.channelActive() will be invoked
+            // and initialization will occur there.
+        }
+    }
+
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+        destroy();
+    }
+
+    @Override
+    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
+        // Initialize early if channel is active already.
+        if (ctx.channel().isActive()) {
+            initialize(ctx);
+        }
+        super.channelRegistered(ctx);
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        // This method will be invoked only if this handler was added
+        // before channelActive() event is fired. If a user adds this handler
+        // after the channelActive() event, initialize() will be called by beforeAdd().
+        initialize(ctx);
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        destroy();
+        super.channelInactive(ctx);
+    }
+
+    private void initialize(ChannelHandlerContext ctx) {
+        // Avoid the case where destroy() is called before scheduling timeouts.
+        // See: https://github.com/netty/netty/issues/143
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Initializing autoflush handler. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
+        }
+        switch (state) {
+            case 1:
+            case 2:
+                return;
+        }
+
+        state = 1;
+
+        EventExecutor loop = ctx.executor();
+
+        lastExecutionTime = System.nanoTime();
+        resenderTimeout = loop.schedule(new WriterIdleTimeoutTask(ctx), resenderTimeNanos, TimeUnit.NANOSECONDS);
+    }
+
+    private void destroy() {
+        state = 2;
+
+        if (resenderTimeout != null) {
+            resenderTimeout.cancel(false);
+            resenderTimeout = null;
+        }
+    }
+
+    private void resendNotAcked(ChannelHandlerContext ctx/* , IdleStateEvent evt */) throws Exception {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Flushing idle Netty channel. MqttClientId = {}.", NettyUtils.clientID(ctx.channel()));
+        }
+        ctx.fireUserEventTriggered(new ResendNotAckedPublishes());
+    }
+}

broker/src/main/java/io/moquette/broker/MQTTConnection.java

@@ -0,0 +1,435 @@
+package io.moquette.broker;
+
+import io.moquette.server.netty.AutoFlushHandler;
+import io.moquette.server.netty.NettyUtils;
+import io.moquette.spi.impl.DebugUtils;
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.security.IAuthenticator;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.handler.codec.mqtt.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
+import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
+import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
+import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
+import static io.netty.handler.codec.mqtt.MqttQoS.*;
+
+final class MQTTConnection {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MQTTConnection.class);
+
+    final Channel channel;
+    private BrokerConfiguration brokerConfig;
+    private IAuthenticator authenticator;
+    private SessionRegistry sessionRegistry;
+    private final PostOffice postOffice;
+    private boolean connected;
+    private final AtomicInteger lastPacketId = new AtomicInteger(0);
+
+    MQTTConnection(Channel channel, BrokerConfiguration brokerConfig, IAuthenticator authenticator,
+                   SessionRegistry sessionRegistry, PostOffice postOffice) {
+        this.channel = channel;
+        this.brokerConfig = brokerConfig;
+        this.authenticator = authenticator;
+        this.sessionRegistry = sessionRegistry;
+        this.postOffice = postOffice;
+        this.connected = false;
+    }
+
+    void handleMessage(MqttMessage msg) {
+        MqttMessageType messageType = msg.fixedHeader().messageType();
+        LOG.debug("Received MQTT message, type: {}, channel: {}", messageType, channel);
+        switch (messageType) {
+            case CONNECT:
+                processConnect((MqttConnectMessage) msg);
+                break;
+            case SUBSCRIBE:
+                processSubscribe((MqttSubscribeMessage) msg);
+                break;
+            case UNSUBSCRIBE:
+                processUnsubscribe((MqttUnsubscribeMessage) msg);
+                break;
+            case PUBLISH:
+                processPublish((MqttPublishMessage) msg);
+                break;
+            case PUBREC:
+                processPubRec(msg);
+                break;
+            case PUBCOMP:
+                processPubComp(msg);
+                break;
+            case PUBREL:
+                processPubRel(msg);
+                break;
+            case DISCONNECT:
+                processDisconnect(msg);
+                break;
+            case PUBACK:
+                processPubAck(msg);
+                break;
+            case PINGREQ:
+                MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP,false, AT_MOST_ONCE,
+                                                                false,0);
+                MqttMessage pingResp = new MqttMessage(pingHeader);
+                channel.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE);
+                break;
+            default:
+                LOG.error("Unknown MessageType: {}, channel: {}", messageType, channel);
+                break;
+        }
+    }
+
+    private void processPubComp(MqttMessage msg) {
+        final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
+        final Session session = sessionRegistry.retrieve(getClientId());
+        session.processPubComp(messageID);
+    }
+
+    private void processPubRec(MqttMessage msg) {
+        final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
+        final Session session = sessionRegistry.retrieve(getClientId());
+        session.processPubRec(messageID);
+    }
+
+    static MqttMessage pubrel(int messageID) {
+        MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0);
+        return new MqttMessage(pubRelHeader, from(messageID));
+    }
+
+    private void processPubAck(MqttMessage msg) {
+        final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
+        Session session = sessionRegistry.retrieve(getClientId());
+        session.pubAckReceived(messageID);
+    }
+
+    void processConnect(MqttConnectMessage msg) {
+        MqttConnectPayload payload = msg.payload();
+        String clientId = payload.clientIdentifier();
+        final String username = payload.userName();
+        LOG.trace("Processing CONNECT message. CId={} username: {} channel: {}", clientId, username, channel);
+
+        if (isNotProtocolVersion(msg, MqttVersion.MQTT_3_1) && isNotProtocolVersion(msg, MqttVersion.MQTT_3_1_1)) {
+            LOG.warn("MQTT protocol version is not valid. CId={} channel: {}", clientId, channel);
+            abortConnection(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION);
+            return;
+        }
+        final boolean cleanSession = msg.variableHeader().isCleanSession();
+        if (clientId == null || clientId.length() == 0) {
+            if (!brokerConfig.isAllowZeroByteClientId()) {
+                LOG.warn("Broker doesn't permit MQTT client ID empty. Username={}, channel: {}", username, channel);
+                abortConnection(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
+                return;
+            }
+
+            if (!cleanSession) {
+                LOG.warn("MQTT client ID cannot be empty for persistent session. Username={}, channel: {}", username, channel);
+                abortConnection(CONNECTION_REFUSED_IDENTIFIER_REJECTED);
+                return;
+            }
+
+            // Generating client id.
+            clientId = UUID.randomUUID().toString().replace("-", "");
+            LOG.debug("Client has connected with server generated id={}, username={}, channel: {}", clientId, username, channel);
+        }
+
+        if (!login(msg, clientId)) {
+            abortConnection(CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
+            channel.close().addListener(CLOSE_ON_FAILURE);
+            return;
+        }
+
+        try {
+            LOG.trace("Binding MQTTConnection (channel: {}) to session", channel);
+            sessionRegistry.bindToSession(this, msg, clientId);
+
+            setupInflightResender(channel);
+
+            NettyUtils.clientID(channel, clientId);
+            LOG.trace("CONNACK sent, channel: {}", channel);
+        } catch (SessionCorruptedException scex) {
+            LOG.warn("MQTT session for client ID {} cannot be created, channel: {}", clientId, channel);
+            abortConnection(CONNECTION_REFUSED_SERVER_UNAVAILABLE);
+        }
+    }
+
+    private void setupInflightResender(Channel channel) {
+        channel.pipeline()
+            .addFirst("inflightResender", new InflightResender(5_000, TimeUnit.MILLISECONDS));
+    }
+
+    private boolean isNotProtocolVersion(MqttConnectMessage msg, MqttVersion version) {
+        return msg.variableHeader().version() != version.protocolLevel();
+    }
+
+    private void abortConnection(MqttConnectReturnCode returnCode) {
+        MqttConnAckMessage badProto = connAck(returnCode, false);
+        channel.writeAndFlush(badProto).addListener(FIRE_EXCEPTION_ON_FAILURE);
+        channel.close().addListener(CLOSE_ON_FAILURE);
+    }
+
+    private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) {
+        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE,
+            false, 0);
+        MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent);
+        return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
+    }
+
+    private boolean login(MqttConnectMessage msg, final String clientId) {
+        // handle user authentication
+        if (msg.variableHeader().hasUserName()) {
+            byte[] pwd = null;
+            if (msg.variableHeader().hasPassword()) {
+                pwd = msg.payload().password().getBytes(StandardCharsets.UTF_8);
+            } else if (!brokerConfig.isAllowAnonymous()) {
+                LOG.error("Client didn't supply any password and MQTT anonymous mode is disabled CId={}", clientId);
+                return false;
+            }
+            final String login = msg.payload().userName();
+            if (!authenticator.checkValid(clientId, login, pwd)) {
+                LOG.error("Authenticator has rejected the MQTT credentials CId={}, username={}", clientId, login);
+                return false;
+            }
+            NettyUtils.userName(channel, login);
+        } else if (!brokerConfig.isAllowAnonymous()) {
+            LOG.error("Client didn't supply any credentials and MQTT anonymous mode is disabled. CId={}", clientId);
+            return false;
+        }
+        return true;
+    }
+
+    void handleConnectionLost() {
+        String clientID = NettyUtils.clientID(channel);
+        if (clientID != null && !clientID.isEmpty()) {
+            LOG.info("Notifying connection lost event. CId = {}, channel: {}", clientID, channel);
+            Session session = sessionRegistry.retrieve(clientID);
+            if (session.hasWill()) {
+                postOffice.fireWill(session.getWill());
+            }
+            if (session.isClean()) {
+                sessionRegistry.remove(clientID);
+            } else {
+                sessionRegistry.disconnect(clientID);
+            }
+            connected = false;
+        }
+        channel.close().addListener(CLOSE_ON_FAILURE);
+    }
+
+    void sendConnAck(boolean isSessionAlreadyPresent) {
+        connected = true;
+        final MqttConnAckMessage ackMessage = connAck(CONNECTION_ACCEPTED, isSessionAlreadyPresent);
+        channel.writeAndFlush(ackMessage);
+    }
+
+    void dropConnection() {
+        channel.close();
+    }
+
+    void processDisconnect(MqttMessage msg) {
+        final String clientID = NettyUtils.clientID(channel);
+        LOG.trace("Start DISCONNECT CId={}, channel: {}", clientID, channel);
+        if (!connected) {
+            LOG.info("DISCONNECT received on already closed connection, CId={}, channel: {}", clientID, channel);
+            return;
+        }
+        sessionRegistry.disconnect(clientID);
+        connected = false;
+        channel.close();
+        LOG.trace("Processed DISCONNECT CId={}, channel: {}", clientID, channel);
+    }
+
+    void processSubscribe(MqttSubscribeMessage msg) {
+        final String clientID = NettyUtils.clientID(channel);
+        if (!connected) {
+            LOG.warn("SUBSCRIBE received on already closed connection, CId={}, channel: {}", clientID, channel);
+            dropConnection();
+            return;
+        }
+        postOffice.subscribeClientToTopics(msg, clientID, NettyUtils.userName(channel), this);
+    }
+
+    void sendSubAckMessage(int messageID, MqttSubAckMessage ackMessage) {
+        final String clientId = NettyUtils.clientID(channel);
+        LOG.trace("Sending SUBACK response CId={}, messageId: {}", clientId, messageID);
+        channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE);
+    }
+
+    private void processUnsubscribe(MqttUnsubscribeMessage msg) {
+        List<String> topics = msg.payload().topics();
+        String clientID = NettyUtils.clientID(channel);
+
+        LOG.trace("Processing UNSUBSCRIBE message. CId={}, topics: {}", clientID, topics);
+        postOffice.unsubscribe(topics, this, msg.variableHeader().messageId());
+    }
+
+    void sendUnsubAckMessage(List<String> topics, String clientID, int messageID) {
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_MOST_ONCE,
+            false, 0);
+        MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID));
+
+        LOG.trace("Sending UNSUBACK message. CId={}, messageId: {}, topics: {}", clientID, messageID, topics);
+        channel.writeAndFlush(ackMessage).addListener(FIRE_EXCEPTION_ON_FAILURE);
+        LOG.trace("Client <{}> unsubscribed from topics <{}>", clientID, topics);
+    }
+
+    void processPublish(MqttPublishMessage msg) {
+        final MqttQoS qos = msg.fixedHeader().qosLevel();
+        final String username = NettyUtils.userName(channel);
+        final String topicName = msg.variableHeader().topicName();
+        final String clientId = getClientId();
+        LOG.trace("Processing PUBLISH message. CId={}, topic: {}, messageId: {}, qos: {}", clientId, topicName,
+                  msg.variableHeader().packetId(), qos);
+        ByteBuf payload = msg.payload();
+        final boolean retain = msg.fixedHeader().isRetain();
+        final Topic topic = new Topic(topicName);
+        if (!topic.isValid()) {
+            LOG.debug("Drop connection because of invalid topic format");
+            dropConnection();
+        }
+        switch (qos) {
+            case AT_MOST_ONCE:
+                postOffice.receivedPublishQos0(topic, username, clientId, payload, retain);
+                break;
+            case AT_LEAST_ONCE: {
+                final int messageID = msg.variableHeader().packetId();
+                postOffice.receivedPublishQos1(this, topic, username, payload, messageID, retain, msg);
+                break;
+            }
+            case EXACTLY_ONCE: {
+                final int messageID = msg.variableHeader().packetId();
+                final Session session = sessionRegistry.retrieve(clientId);
+                session.receivedPublishQos2(messageID, msg);
+                postOffice.receivedPublishQos2(this, msg);
+//                msg.release();
+                break;
+            }
+            default:
+                LOG.error("Unknown QoS-Type:{}", qos);
+                break;
+        }
+    }
+
+    void sendPublishReceived(int messageID) {
+        LOG.trace("sendPubRec invoked on channel: {}", channel);
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE,
+            false, 0);
+        MqttPubAckMessage pubRecMessage = new MqttPubAckMessage(fixedHeader, from(messageID));
+        sendIfWritableElseDrop(pubRecMessage);
+    }
+
+    private void processPubRel(MqttMessage msg) {
+        final Session session = sessionRegistry.retrieve(getClientId());
+        final int messageID = ((MqttMessageIdVariableHeader) msg.variableHeader()).messageId();
+        session.receivedPubRelQos2(messageID);
+        sendPubCompMessage(messageID);
+    }
+
+    void sendPublish(MqttPublishMessage publishMsg) {
+        final int packetId = publishMsg.variableHeader().packetId();
+        final String topicName = publishMsg.variableHeader().topicName();
+        final String clientId = getClientId();
+        MqttQoS qos = publishMsg.fixedHeader().qosLevel();
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("Sending PUBLISH({}) message. MessageId={}, CId={}, topic={}, payload={}", qos, packetId,
+                      clientId, topicName, DebugUtils.payload2Str(publishMsg.payload()));
+        } else {
+            LOG.debug("Sending PUBLISH({}) message. MessageId={}, CId={}, topic={}", qos, packetId, clientId,
+                      topicName);
+        }
+        sendIfWritableElseDrop(publishMsg);
+    }
+
+    void sendIfWritableElseDrop(MqttMessage msg) {
+        if (channel.isWritable()) {
+            channel.write(msg);
+        }
+    }
+
+    public void writabilityChanged() {
+        if (channel.isWritable()) {
+            LOG.debug("Channel {} is again writable", channel);
+            final Session session = sessionRegistry.retrieve(getClientId());
+            session.writabilityChanged();
+        }
+    }
+
+    void sendPubAck(int messageID) {
+        LOG.trace("sendPubAck invoked");
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, AT_MOST_ONCE,
+                                                  false, 0);
+        MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(fixedHeader, from(messageID));
+        sendIfWritableElseDrop(pubAckMessage);
+    }
+
+    private void sendPubCompMessage(int messageID) {
+        LOG.trace("Sending PUBCOMP message on channel: {}, messageId: {}", channel, messageID);
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, AT_MOST_ONCE, false, 0);
+        MqttMessage pubCompMessage = new MqttMessage(fixedHeader, from(messageID));
+        sendIfWritableElseDrop(pubCompMessage);
+    }
+
+    String getClientId() {
+        return NettyUtils.clientID(channel);
+    }
+
+    public void sendPublishRetainedQos0(Topic topic, MqttQoS qos, ByteBuf payload) {
+        MqttPublishMessage publishMsg = retainedPublish(topic.toString(), qos, payload);
+        sendPublish(publishMsg);
+    }
+
+    public void sendPublishRetainedWithPacketId(Topic topic, MqttQoS qos, ByteBuf payload) {
+        final int packetId = nextPacketId();
+        MqttPublishMessage publishMsg = retainedPublishWithMessageId(topic.toString(), qos, payload, packetId);
+        sendPublish(publishMsg);
+    }
+
+    private static MqttPublishMessage retainedPublish(String topic, MqttQoS qos, ByteBuf message) {
+        return retainedPublishWithMessageId(topic, qos, message, 0);
+    }
+
+    private static MqttPublishMessage retainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message,
+                                                                   int messageId) {
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, true, 0);
+        MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
+        return new MqttPublishMessage(fixedHeader, varHeader, message);
+    }
+
+    // TODO move this method in Session
+    void sendPublishNotRetainedQos0(Topic topic, MqttQoS qos, ByteBuf payload) {
+        MqttPublishMessage publishMsg = notRetainedPublish(topic.toString(), qos, payload);
+        sendPublish(publishMsg);
+    }
+
+    static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) {
+        return notRetainedPublishWithMessageId(topic, qos, message, 0);
+    }
+
+    static MqttPublishMessage notRetainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message,
+                                                              int messageId) {
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
+        MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
+        return new MqttPublishMessage(fixedHeader, varHeader, message);
+    }
+
+    public void resendNotAckedPublishes() {
+        final Session session = sessionRegistry.retrieve(getClientId());
+        session.resendInflightNotAcked();
+    }
+
+    int nextPacketId() {
+        return lastPacketId.incrementAndGet();
+    }
+
+    @Override
+    public String toString() {
+        return "MQTTConnection{channel=" + channel + ", connected=" + connected + '}';
+    }
+}

broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java

@@ -0,0 +1,24 @@
+package io.moquette.broker;
+
+import io.moquette.spi.security.IAuthenticator;
+import io.netty.channel.Channel;
+
+class MQTTConnectionFactory {
+
+    private final BrokerConfiguration brokerConfig;
+    private final IAuthenticator authenticator;
+    private final SessionRegistry sessionRegistry;
+    private final PostOffice postOffice;
+
+    MQTTConnectionFactory(BrokerConfiguration brokerConfig, IAuthenticator authenticator,
+                          SessionRegistry sessionRegistry, PostOffice postOffice) {
+        this.brokerConfig = brokerConfig;
+        this.authenticator = authenticator;
+        this.sessionRegistry = sessionRegistry;
+        this.postOffice = postOffice;
+    }
+
+    public MQTTConnection create(Channel channel) {
+        return new MQTTConnection(channel, brokerConfig, authenticator, sessionRegistry, postOffice);
+    }
+}

broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java

@@ -0,0 +1,35 @@
+package io.moquette.broker;
+
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/*
+* In memory retained messages store
+* */
+final class MemoryRetainedRepository implements IRetainedRepository {
+
+    private final ConcurrentMap<String, MqttPublishMessage> storage = new ConcurrentHashMap<>();
+
+    @Override
+    public void cleanRetained(Topic topic) {
+        storage.remove(topic.toString());
+    }
+
+    @Override
+    public void retain(Topic topic, MqttPublishMessage msg) {
+        storage.put(topic.toString(), msg);
+    }
+
+    @Override
+    public boolean isEmtpy() {
+        return storage.isEmpty();
+    }
+
+    @Override
+    public MqttPublishMessage retainedOnTopic(String topic) {
+        return storage.get(topic);
+    }
+}

broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java

@@ -0,0 +1,383 @@
+/*
+ * Copyright (c) 2012-2018 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.moquette.broker;
+
+import io.moquette.BrokerConstants;
+import io.moquette.server.config.IConfig;
+import io.moquette.server.netty.BugSnagErrorsHandler;
+import io.moquette.server.netty.MoquetteIdleTimeoutHandler;
+import io.moquette.server.netty.metrics.*;
+import io.moquette.spi.security.ISslContextCreator;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.*;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import io.netty.handler.codec.MessageToMessageEncoder;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import io.netty.handler.codec.mqtt.MqttDecoder;
+import io.netty.handler.codec.mqtt.MqttEncoder;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.concurrent.Future;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+import static io.moquette.BrokerConstants.*;
+import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
+
+public class NewNettyAcceptor {
+
+    private static final String MQTT_SUBPROTOCOL_CSV_LIST = "mqtt, mqttv3.1, mqttv3.1.1";
+
+    static class WebSocketFrameToByteBufDecoder extends MessageToMessageDecoder<BinaryWebSocketFrame> {
+
+        @Override
+        protected void decode(ChannelHandlerContext chc, BinaryWebSocketFrame frame, List<Object> out)
+                throws Exception {
+            // convert the frame to a ByteBuf
+            ByteBuf bb = frame.content();
+            // System.out.println("WebSocketFrameToByteBufDecoder decode - " +
+            // ByteBufUtil.hexDump(bb));
+            bb.retain();
+            out.add(bb);
+        }
+    }
+
+    static class ByteBufToWebSocketFrameEncoder extends MessageToMessageEncoder<ByteBuf> {
+
+        @Override
+        protected void encode(ChannelHandlerContext chc, ByteBuf bb, List<Object> out) throws Exception {
+            // convert the ByteBuf to a WebSocketFrame
+            BinaryWebSocketFrame result = new BinaryWebSocketFrame();
+            // System.out.println("ByteBufToWebSocketFrameEncoder encode - " +
+            // ByteBufUtil.hexDump(bb));
+            result.content().writeBytes(bb);
+            out.add(result);
+        }
+    }
+
+    private abstract static class PipelineInitializer {
+
+        abstract void init(SocketChannel channel) throws Exception;
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(NewNettyAcceptor.class);
+
+    EventLoopGroup m_bossGroup;
+    EventLoopGroup m_workerGroup;
+    BytesMetricsCollector m_bytesMetricsCollector = new BytesMetricsCollector();
+    MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
+    private Optional<? extends ChannelInboundHandler> metrics;
+    private Optional<? extends ChannelInboundHandler> errorsCather;
+
+    private int nettySoBacklog;
+    private boolean nettySoReuseaddr;
+    private boolean nettyTcpNodelay;
+    private boolean nettySoKeepalive;
+    private int nettyChannelTimeoutSeconds;
+    private int maxBytesInMessage;
+
+    private Class<? extends ServerSocketChannel> channelClass;
+
+    public void initialize(NewNettyMQTTHandler mqttHandler, IConfig props, ISslContextCreator sslCtxCreator) {
+        LOG.debug("Initializing Netty acceptor");
+
+        nettySoBacklog = props.intProp(BrokerConstants.NETTY_SO_BACKLOG_PROPERTY_NAME, 128);
+        nettySoReuseaddr = props.boolProp(BrokerConstants.NETTY_SO_REUSEADDR_PROPERTY_NAME, true);
+        nettyTcpNodelay = props.boolProp(BrokerConstants.NETTY_TCP_NODELAY_PROPERTY_NAME, true);
+        nettySoKeepalive = props.boolProp(BrokerConstants.NETTY_SO_KEEPALIVE_PROPERTY_NAME, true);
+        nettyChannelTimeoutSeconds = props.intProp(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, 10);
+        maxBytesInMessage = props.intProp(BrokerConstants.NETTY_MAX_BYTES_PROPERTY_NAME,
+                BrokerConstants.DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE);
+
+        boolean epoll = props.boolProp(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, false);
+        if (epoll) {
+            LOG.info("Netty is using Epoll");
+            m_bossGroup = new EpollEventLoopGroup();
+            m_workerGroup = new EpollEventLoopGroup();
+            channelClass = EpollServerSocketChannel.class;
+        } else {
+            LOG.info("Netty is using NIO");
+            m_bossGroup = new NioEventLoopGroup();
+            m_workerGroup = new NioEventLoopGroup();
+            channelClass = NioServerSocketChannel.class;
+        }
+
+        final boolean useFineMetrics = props.boolProp(METRICS_ENABLE_PROPERTY_NAME, false);
+        if (useFineMetrics) {
+            DropWizardMetricsHandler metricsHandler = new DropWizardMetricsHandler();
+            metricsHandler.init(props);
+            this.metrics = Optional.of(metricsHandler);
+        } else {
+            this.metrics = Optional.empty();
+        }
+
+        final boolean useBugSnag = props.boolProp(BUGSNAG_ENABLE_PROPERTY_NAME, false);
+        if (useBugSnag) {
+            BugSnagErrorsHandler bugSnagHandler = new BugSnagErrorsHandler();
+            bugSnagHandler.init(props);
+            this.errorsCather = Optional.of(bugSnagHandler);
+        } else {
+            this.errorsCather = Optional.empty();
+        }
+        initializePlainTCPTransport(mqttHandler, props);
+//        initializeWebSocketTransport(mqttHandler, props);
+//        String sslTcpPortProp = props.getProperty(BrokerConstants.SSL_PORT_PROPERTY_NAME);
+//        String wssPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
+//        if (sslTcpPortProp != null || wssPortProp != null) {
+//            SslContext sslContext = sslCtxCreator.initSSLContext();
+//            if (sslContext == null) {
+//                LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
+//                return;
+//            }
+//            initializeSSLTCPTransport(mqttHandler, props, sslContext);
+//            initializeWSSTransport(mqttHandler, props, sslContext);
+//        }
+    }
+
+    private void initFactory(String host, int port, String protocol, final PipelineInitializer pipeliner) {
+        LOG.debug("Initializing server. Protocol={}", protocol);
+        ServerBootstrap b = new ServerBootstrap();
+        b.group(m_bossGroup, m_workerGroup).channel(channelClass)
+                .childHandler(new ChannelInitializer<SocketChannel>() {
+
+                    @Override
+                    public void initChannel(SocketChannel ch) throws Exception {
+                        pipeliner.init(ch);
+                    }
+                })
+                .option(ChannelOption.SO_BACKLOG, nettySoBacklog)
+                .option(ChannelOption.SO_REUSEADDR, nettySoReuseaddr)
+                .childOption(ChannelOption.TCP_NODELAY, nettyTcpNodelay)
+                .childOption(ChannelOption.SO_KEEPALIVE, nettySoKeepalive);
+        try {
+            LOG.debug("Binding server. host={}, port={}", host, port);
+            // Bind and start to accept incoming connections.
+            ChannelFuture f = b.bind(host, port);
+            LOG.info("Server bound to host={}, port={}, protocol={}", host, port, protocol);
+            f.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);
+        } catch (InterruptedException ex) {
+            LOG.error("An interruptedException was caught while initializing server. Protocol={}", protocol, ex);
+        }
+    }
+
+    private void initializePlainTCPTransport(NewNettyMQTTHandler handler, IConfig props) {
+        LOG.debug("Configuring TCP MQTT transport");
+        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+        String tcpPortProp = props.getProperty(PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+        if (DISABLED_PORT_BIND.equals(tcpPortProp)) {
+            LOG.info("Property {} has been set to {}. TCP MQTT will be disabled", BrokerConstants.PORT_PROPERTY_NAME,
+                    DISABLED_PORT_BIND);
+            return;
+        }
+        int port = Integer.parseInt(tcpPortProp);
+        initFactory(host, port, "TCP MQTT", new PipelineInitializer() {
+
+            @Override
+            void init(SocketChannel channel) {
+                ChannelPipeline pipeline = channel.pipeline();
+                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+                // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
+                if (errorsCather.isPresent()) {
+                    pipeline.addLast("bugsnagCatcher", errorsCather.get());
+                }
+                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+                if (metrics.isPresent()) {
+                    pipeline.addLast("wizardMetrics", metrics.get());
+                }
+                pipeline.addLast("handler", handler);
+            }
+        });
+    }
+
+//    private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) {
+//        LOG.debug("Configuring Websocket MQTT transport");
+//        String webSocketPortProp = props.getProperty(WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+//        if (DISABLED_PORT_BIND.equals(webSocketPortProp)) {
+//            // Do nothing no WebSocket configured
+//            LOG.info("Property {} has been setted to {}. Websocket MQTT will be disabled",
+//                    BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+//            return;
+//        }
+//        int port = Integer.parseInt(webSocketPortProp);
+//
+//        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+//
+//        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+//        initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {
+//
+//            @Override
+//            void init(SocketChannel channel) {
+//                ChannelPipeline pipeline = channel.pipeline();
+//                pipeline.addLast(new HttpServerCodec());
+//                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+//                pipeline.addLast("webSocketHandler",
+//                        new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
+//                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
+//                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
+//                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+//                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+//                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+//                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+//                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+//                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+//                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+//                pipeline.addLast("handler", handler);
+//            }
+//        });
+//    }
+//
+//    private void initializeSSLTCPTransport(NettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+//        LOG.debug("Configuring SSL MQTT transport");
+//        String sslPortProp = props.getProperty(SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+//        if (DISABLED_PORT_BIND.equals(sslPortProp)) {
+//            // Do nothing no SSL configured
+//            LOG.info("Property {} has been set to {}. SSL MQTT will be disabled",
+//                    BrokerConstants.SSL_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+//            return;
+//        }
+//
+//        int sslPort = Integer.parseInt(sslPortProp);
+//        LOG.debug("Starting SSL on port {}", sslPort);
+//
+//        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+//        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+//        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
+//        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+//        initFactory(host, sslPort, "SSL MQTT", new PipelineInitializer() {
+//
+//            @Override
+//            void init(SocketChannel channel) throws Exception {
+//                ChannelPipeline pipeline = channel.pipeline();
+//                pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
+//                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+//                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+//                // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
+//                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+//                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+//                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+//                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+//                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+//                pipeline.addLast("handler", handler);
+//            }
+//        });
+//    }
+//
+//    private void initializeWSSTransport(NettyMQTTHandler handler, IConfig props, SslContext sslContext) {
+//        LOG.debug("Configuring secure websocket MQTT transport");
+//        String sslPortProp = props.getProperty(WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+//        if (DISABLED_PORT_BIND.equals(sslPortProp)) {
+//            // Do nothing no SSL configured
+//            LOG.info("Property {} has been set to {}. Secure websocket MQTT will be disabled",
+//                    BrokerConstants.WSS_PORT_PROPERTY_NAME, DISABLED_PORT_BIND);
+//            return;
+//        }
+//        int sslPort = Integer.parseInt(sslPortProp);
+//        final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
+//        String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
+//        String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
+//        final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
+//        initFactory(host, sslPort, "Secure websocket", new PipelineInitializer() {
+//
+//            @Override
+//            void init(SocketChannel channel) throws Exception {
+//                ChannelPipeline pipeline = channel.pipeline();
+//                pipeline.addLast("ssl", createSslHandler(channel, sslContext, needsClientAuth));
+//                pipeline.addLast("httpEncoder", new HttpResponseEncoder());
+//                pipeline.addLast("httpDecoder", new HttpRequestDecoder());
+//                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
+//                pipeline.addLast("webSocketHandler",
+//                        new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
+//                pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
+//                pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
+//                pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0));
+//                pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
+//                pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
+//                pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage));
+//                pipeline.addLast("encoder", MqttEncoder.INSTANCE);
+//                pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
+//                pipeline.addLast("messageLogger", new MQTTMessageLogger());
+//                pipeline.addLast("handler", handler);
+//            }
+//        });
+//    }
+
+    @SuppressWarnings("FutureReturnValueIgnored")
+    public void close() {
+        LOG.debug("Closing Netty acceptor...");
+        if (m_workerGroup == null || m_bossGroup == null) {
+            LOG.error("Netty acceptor is not initialized");
+            throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
+        }
+        Future<?> workerWaiter = m_workerGroup.shutdownGracefully();
+        Future<?> bossWaiter = m_bossGroup.shutdownGracefully();
+
+        /*
+         * We shouldn't raise an IllegalStateException if we are interrupted. If we did so, the
+         * broker is not shut down properly.
+         */
+        LOG.info("Waiting for worker and boss event loop groups to terminate...");
+        try {
+            workerWaiter.await(10, TimeUnit.SECONDS);
+            bossWaiter.await(10, TimeUnit.SECONDS);
+        } catch (InterruptedException iex) {
+            LOG.warn("An InterruptedException was caught while waiting for event loops to terminate...");
+        }
+
+        if (!m_workerGroup.isTerminated()) {
+            LOG.warn("Forcing shutdown of worker event loop...");
+            m_workerGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
+        }
+
+        if (!m_bossGroup.isTerminated()) {
+            LOG.warn("Forcing shutdown of boss event loop...");
+            m_bossGroup.shutdownGracefully(0L, 0L, TimeUnit.MILLISECONDS);
+        }
+
+        MessageMetrics metrics = m_metricsCollector.computeMetrics();
+        BytesMetrics bytesMetrics = m_bytesMetricsCollector.computeMetrics();
+        LOG.info("Metrics messages[read={}, write={}] bytes[read={}, write={}]", metrics.messagesRead(),
+                metrics.messagesWrote(), bytesMetrics.readBytes(), bytesMetrics.wroteBytes());
+    }
+
+//    private ChannelHandler createSslHandler(SocketChannel channel, SslContext sslContext, boolean needsClientAuth) {
+//        SSLEngine sslEngine = sslContext.newEngine(
+//                channel.alloc(),
+//                channel.remoteAddress().getHostString(),
+//                channel.remoteAddress().getPort());
+//        sslEngine.setUseClientMode(false);
+//        if (needsClientAuth) {
+//            sslEngine.setNeedClientAuth(true);
+//        }
+//        return new SslHandler(sslEngine);
+//    }
+}

broker/src/main/java/io/moquette/broker/NewNettyMQTTHandler.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright (c) 2012-2018 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.moquette.broker;
+
+import io.moquette.server.netty.NettyUtils;
+import io.netty.channel.*;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.util.AttributeKey;
+import io.netty.util.ReferenceCountUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
+
+@Sharable
+public class NewNettyMQTTHandler extends ChannelInboundHandlerAdapter {
+
+    private static final Logger LOG = LoggerFactory.getLogger(NewNettyMQTTHandler.class);
+
+    private static final String ATTR_CONNECTION = "connection";
+    private static final AttributeKey<Object> ATTR_KEY_CONNECTION = AttributeKey.valueOf(ATTR_CONNECTION);
+
+    private MQTTConnectionFactory connectionFactory;
+
+    NewNettyMQTTHandler(MQTTConnectionFactory connectionFactory) {
+        this.connectionFactory = connectionFactory;
+    }
+
+    private static void mqttConnection(Channel channel, MQTTConnection connection) {
+        channel.attr(ATTR_KEY_CONNECTION).set(connection);
+    }
+
+    private static MQTTConnection mqttConnection(Channel channel) {
+        return (MQTTConnection) channel.attr(ATTR_KEY_CONNECTION).get();
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
+        MqttMessage msg = (MqttMessage) message;
+        if (msg.fixedHeader() == null) {
+            throw new IOException("Unknown packet");
+        }
+        final MQTTConnection mqttConnection = mqttConnection(ctx.channel());
+        try {
+            mqttConnection.handleMessage(msg);
+        } catch (Throwable ex) {
+            //ctx.fireExceptionCaught(ex);
+            LOG.error("Error processing protocol message: {}", msg.fixedHeader().messageType(), ex);
+            ctx.channel().close().addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture future) {
+                    LOG.info("Closed client channel due to exception in processing");
+                }
+            });
+        } finally {
+            ReferenceCountUtil.release(msg);
+        }
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) {
+        MQTTConnection connection = connectionFactory.create(ctx.channel());
+        mqttConnection(ctx.channel(), connection);
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) {
+        final MQTTConnection mqttConnection = mqttConnection(ctx.channel());
+        mqttConnection.handleConnectionLost();
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        LOG.error("Unexpected exception while processing MQTT message. Closing Netty channel. CId={}",
+                  NettyUtils.clientID(ctx.channel()), cause);
+        ctx.close().addListener(CLOSE_ON_FAILURE);
+    }
+
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) {
+//        if (ctx.channel().isWritable()) {
+//            m_processor.notifyChannelWritable(ctx.channel());
+//        }
+        final MQTTConnection mqttConnection = mqttConnection(ctx.channel());
+        mqttConnection.writabilityChanged();
+        ctx.fireChannelWritabilityChanged();
+    }
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof InflightResender.ResendNotAckedPublishes) {
+            final MQTTConnection mqttConnection = mqttConnection(ctx.channel());
+            mqttConnection.resendNotAckedPublishes();
+        }
+        ctx.fireUserEventTriggered(evt);
+    }
+
+}

broker/src/main/java/io/moquette/broker/PostOffice.java

@@ -0,0 +1,274 @@
+package io.moquette.broker;
+
+import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
+import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.security.IAuthorizatorPolicy;
+import io.netty.buffer.ByteBuf;
+import io.netty.handler.codec.mqtt.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static io.moquette.spi.impl.Utils.messageId;
+import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
+import static io.netty.handler.codec.mqtt.MqttQoS.*;
+
+class PostOffice {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PostOffice.class);
+
+    private final Authorizator authorizator;
+    private final ISubscriptionsDirectory subscriptions;
+    private final IRetainedRepository retainedRepository;
+    private SessionRegistry sessionRegistry;
+
+    PostOffice(ISubscriptionsDirectory subscriptions, IAuthorizatorPolicy authorizatorPolicy,
+               IRetainedRepository retainedRepository, SessionRegistry sessionRegistry) {
+        this.authorizator = new Authorizator(authorizatorPolicy);
+        this.subscriptions = subscriptions;
+        this.retainedRepository = retainedRepository;
+        this.sessionRegistry = sessionRegistry;
+    }
+
+    public void init(SessionRegistry sessionRegistry) {
+        this.sessionRegistry = sessionRegistry;
+    }
+
+    public void fireWill(Session.Will will) {
+        // MQTT 3.1.2.8-17
+        publish2Subscribers(will.payload, new Topic(will.topic), will.qos);
+    }
+
+    public void subscribeClientToTopics(MqttSubscribeMessage msg, String clientID, String username,
+                                        MQTTConnection mqttConnection) {
+        // verify which topics of the subscribe ongoing has read access permission
+        int messageID = messageId(msg);
+        List<MqttTopicSubscription> ackTopics = authorizator.verifyTopicsReadAccess(clientID, username, msg);
+        MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID);
+
+        // store topics subscriptions in session
+        List<Subscription> newSubscriptions = ackTopics.stream()
+            .filter(req -> req.qualityOfService() != FAILURE)
+            .map(req -> {
+                final Topic topic = new Topic(req.topicName());
+                return new Subscription(clientID, topic, req.qualityOfService());
+            }).collect(Collectors.toList());
+
+        for (Subscription subscription : newSubscriptions) {
+            subscriptions.add(subscription);
+        }
+
+        // add the subscriptions to Session
+        Session session = sessionRegistry.retrieve(clientID);
+        session.addSubscriptions(newSubscriptions);
+
+        // send ack message
+        mqttConnection.sendSubAckMessage(messageID, ackMessage);
+
+        publishRetainedMessagesForSubscriptions(clientID, newSubscriptions);
+
+        // TODO notify the Observables
+//        for (Subscription subscription : newSubscriptions) {
+//            m_interceptor.notifyTopicSubscribed(newSubscription, username);
+//        }
+    }
+
+    private void publishRetainedMessagesForSubscriptions(String clientID, List<Subscription> newSubscriptions) {
+        Session targetSession = this.sessionRegistry.retrieve(clientID);
+        for (Subscription subscription : newSubscriptions) {
+            final String topicFilter = subscription.getTopicFilter().toString();
+            final MqttPublishMessage retainedMsg = retainedRepository.retainedOnTopic(topicFilter);
+
+            if (retainedMsg == null) {
+                // not found
+                continue;
+            }
+
+            final MqttQoS retainedQos = retainedMsg.fixedHeader().qosLevel();
+            MqttQoS qos = lowerQosToTheSubscriptionDesired(subscription, retainedQos);
+
+            final ByteBuf origPayload = retainedMsg.payload();
+            ByteBuf payload = origPayload.retainedDuplicate();
+            targetSession.sendRetainedPublishOnSessionAtQos(subscription.getTopicFilter(), qos, payload);
+        }
+    }
+
+    /**
+     * Create the SUBACK response from a list of topicFilters
+     */
+    private MqttSubAckMessage doAckMessageFromValidateFilters(List<MqttTopicSubscription> topicFilters, int messageId) {
+        List<Integer> grantedQoSLevels = new ArrayList<>();
+        for (MqttTopicSubscription req : topicFilters) {
+            grantedQoSLevels.add(req.qualityOfService().value());
+        }
+
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_MOST_ONCE,
+            false, 0);
+        MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);
+        return new MqttSubAckMessage(fixedHeader, from(messageId), payload);
+    }
+
+    public void unsubscribe(List<String> topics, MQTTConnection mqttConnection, int messageId) {
+        final String clientID = mqttConnection.getClientId();
+        for (String t : topics) {
+            Topic topic = new Topic(t);
+            boolean validTopic = topic.isValid();
+            if (!validTopic) {
+                // close the connection, not valid topicFilter is a protocol violation
+                mqttConnection.dropConnection();
+                LOG.warn("Topic filter is not valid. CId={}, topics: {}, offending topic filter: {}", clientID,
+                         topics, topic);
+                return;
+            }
+
+            LOG.trace("Removing subscription. CId={}, topic={}", clientID, topic);
+            subscriptions.removeSubscription(topic, clientID);
+
+            // TODO remove the subscriptions to Session
+//            clientSession.unsubscribeFrom(topic);
+
+            //TODO notify interceptors
+//            String username = NettyUtils.userName(channel);
+//            m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username);
+        }
+
+        // ack the client
+        mqttConnection.sendUnsubAckMessage(topics, clientID, messageId);
+    }
+
+    void receivedPublishQos0(Topic topic, String username, String clientID, ByteBuf payload, boolean retain) {
+        if (!authorizator.canWrite(topic, username, clientID)) {
+            LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic: {}", clientID, topic);
+            return;
+        }
+        publish2Subscribers(payload, topic, AT_MOST_ONCE);
+
+        if (retain) {
+            // QoS == 0 && retain => clean old retained
+            retainedRepository.cleanRetained(topic);
+        }
+// TODO
+//        m_interceptor.notifyTopicPublished(msg, clientID, username);
+    }
+
+    void receivedPublishQos1(MQTTConnection connection, Topic topic, String username, ByteBuf payload, int messageID,
+                             boolean retain, MqttPublishMessage msg) {
+        // verify if topic can be write
+        topic.getTokens();
+        if (!topic.isValid()) {
+            LOG.warn("Invalid topic format, force close the connection");
+            connection.dropConnection();
+            return;
+        }
+        final String clientId = connection.getClientId();
+        if (!authorizator.canWrite(topic, username, clientId)) {
+            LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic: {}", clientId, topic);
+            return;
+        }
+
+        publish2Subscribers(payload, topic, AT_LEAST_ONCE);
+
+        connection.sendPubAck(messageID);
+
+        if (retain) {
+            if (!payload.isReadable()) {
+                retainedRepository.cleanRetained(topic);
+            } else {
+                // before wasn't stored
+                retainedRepository.retain(topic, msg);
+            }
+        }
+//TODO
+//        m_interceptor.notifyTopicPublished(msg, clientID, username);
+    }
+
+    private void publish2Subscribers(ByteBuf origPayload, Topic topic, MqttQoS publishingQos) {
+        Set<Subscription> topicMatchingSubscriptions = subscriptions.matchQosSharpening(topic);
+
+        for (final Subscription sub : topicMatchingSubscriptions) {
+            MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
+            Session targetSession = this.sessionRegistry.retrieve(sub.getClientId());
+
+            boolean isSessionPresent = targetSession != null;
+            if (isSessionPresent) {
+                LOG.debug("Sending PUBLISH message to active subscriber CId: {}, topicFilter: {}, qos: {}",
+                          sub.getClientId(), sub.getTopicFilter(), qos);
+                // we need to retain because duplicate only copy r/w indexes and don't retain() causing refCnt = 0
+                ByteBuf payload = origPayload.retainedDuplicate();
+                targetSession.sendPublishOnSessionAtQos(topic, qos, payload);
+            } else {
+                // If we are, the subscriber disconnected after the subscriptions tree selected that session as a
+                // destination.
+                LOG.debug("PUBLISH to not yet present session. CId: {}, topicFilter: {}, qos: {}", sub.getClientId(),
+                          sub.getTopicFilter(), qos);
+            }
+        }
+    }
+
+    /**
+     * First phase of a publish QoS2 protocol, sent by publisher to the broker. Publish to all interested
+     * subscribers.
+     */
+    void receivedPublishQos2(MQTTConnection connection, MqttPublishMessage mqttPublishMessage) {
+        LOG.trace("Processing PUBREL message on connection: {}", connection);
+        final Topic topic = new Topic(mqttPublishMessage.variableHeader().topicName());
+        final ByteBuf payload = mqttPublishMessage.payload();
+        publish2Subscribers(payload, topic, EXACTLY_ONCE);
+
+        final boolean retained = mqttPublishMessage.fixedHeader().isRetain();
+        if (retained) {
+            if (!payload.isReadable()) {
+                retainedRepository.cleanRetained(topic);
+            } else {
+                // before wasn't stored
+                retainedRepository.retain(topic, mqttPublishMessage);
+            }
+        }
+
+        //TODO here we should notify to the listeners
+        //m_interceptor.notifyTopicPublished(msg, clientID, username);
+    }
+
+    static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) {
+        if (qos.value() > sub.getRequestedQos().value()) {
+            qos = sub.getRequestedQos();
+        }
+        return qos;
+    }
+
+    /**
+     * Intended usage is only for embedded versions of the broker, where the hosting application
+     * want to use the broker to send a publish message. Like normal external publish message but
+     * with some changes to avoid security check, and the handshake phases for Qos1 and Qos2. It
+     * also doesn't notifyTopicPublished because using internally the owner should already know
+     * where it's publishing.
+     *
+     * @param msg
+     *            the message to publish.
+     * @param clientId
+     *            the clientID
+     */
+    public void internalPublish(MqttPublishMessage msg, final String clientId) {
+        final MqttQoS qos = msg.fixedHeader().qosLevel();
+        final Topic topic = new Topic(msg.variableHeader().topicName());
+        final ByteBuf payload = msg.payload();
+        LOG.info("Sending internal PUBLISH message Topic={}, qos={}", topic, qos);
+
+        publish2Subscribers(payload, topic, qos);
+
+        if (!msg.fixedHeader().isRetain()) {
+            return;
+        }
+        if (qos == AT_MOST_ONCE || msg.payload().readableBytes() == 0) {
+            // QoS == 0 && retain => clean old retained
+            retainedRepository.cleanRetained(topic);
+            return;
+        }
+        retainedRepository.retain(topic, msg);
+    }
+}

broker/src/main/java/io/moquette/broker/Server.java

@@ -0,0 +1,142 @@
+package io.moquette.broker;
+
+import io.moquette.BrokerConstants;
+import io.moquette.interception.InterceptHandler;
+import io.moquette.persistence.MemoryStorageService;
+import io.moquette.server.config.FileResourceLoader;
+import io.moquette.server.config.IConfig;
+import io.moquette.server.config.IResourceLoader;
+import io.moquette.server.config.ResourceLoaderConfig;
+import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.impl.SessionsRepository;
+import io.moquette.spi.impl.security.AcceptAllAuthenticator;
+import io.moquette.spi.impl.security.PermitAllAuthorizatorPolicy;
+import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
+import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
+import io.moquette.spi.security.IAuthenticator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
+import io.moquette.spi.security.ISslContextCreator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static io.moquette.logging.LoggingUtils.getInterceptorIds;
+
+public class Server {
+
+    private static final Logger LOG = LoggerFactory.getLogger(io.moquette.server.Server.class);
+
+    private ScheduledExecutorService scheduler;
+    private NewNettyAcceptor acceptor;
+    private volatile boolean initialized;
+
+    public static void main(String[] args) throws IOException {
+        final Server server = new Server();
+        server.startServer();
+        System.out.println("Server started, version 0.12-SNAPSHOT");
+        //Bind  a shutdown hook
+        Runtime.getRuntime().addShutdownHook(new Thread(server::stopServer));
+    }
+
+    /**
+     * Starts Moquette bringing the configuration from the file located at m_config/moquette.conf
+     *
+     * @throws IOException in case of any IO error.
+     */
+    public void startServer() throws IOException {
+        File defaultConfigurationFile = defaultConfigFile();
+        LOG.info("Starting Moquette server. Configuration file path={}", defaultConfigurationFile.getAbsolutePath());
+        IResourceLoader filesystemLoader = new FileResourceLoader(defaultConfigurationFile);
+        final IConfig config = new ResourceLoaderConfig(filesystemLoader);
+        startServer(config);
+    }
+
+    public void startServer(IConfig config) throws IOException {
+        LOG.debug("Starting Moquette server using IConfig instance");
+        startServer(config, null);
+    }
+
+    public void startServer(IConfig config, List<? extends InterceptHandler> handlers) throws IOException {
+        LOG.debug("Starting moquette server using IConfig instance and intercept handlers");
+        startServer(config, handlers, null, null, null);
+    }
+
+    public void startServer(IConfig config, List<? extends InterceptHandler> handlers, ISslContextCreator sslCtxCreator,
+                            IAuthenticator authenticator, IAuthorizatorPolicy authorizatorPolicy) throws IOException {
+        final long start = System.currentTimeMillis();
+        if (handlers == null) {
+            handlers = Collections.emptyList();
+        }
+        LOG.trace("Starting Moquette Server. MQTT message interceptors={}", getInterceptorIds(handlers));
+
+        scheduler = Executors.newScheduledThreadPool(1);
+
+        final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
+        if (handlerProp != null) {
+            config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
+        }
+        final String persistencePath = config.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME);
+        LOG.debug("Configuring Using persistent store file, path={}", persistencePath);
+//        m_processorBootstrapper = new ProtocolProcessorBootstrapper();
+//        final ProtocolProcessor processor = m_processorBootstrapper.init(config, handlers, authenticator, authorizator,
+//            this);
+        LOG.debug("Initialized MQTT protocol processor");
+//        if (sslCtxCreator == null) {
+//            LOG.info("Using default SSL context creator");
+//            sslCtxCreator = new DefaultMoquetteSslContextCreator(config);
+//        }
+        // DBG
+        if (authenticator == null) {
+            authenticator = new AcceptAllAuthenticator();
+        }
+        if (authorizatorPolicy == null) {
+            authorizatorPolicy = new PermitAllAuthorizatorPolicy();
+        }
+
+        MemoryStorageService memStorage = new MemoryStorageService(null, null);
+        ISessionsStore sessionStore = memStorage.sessionsStore();
+        SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null);
+        // DBG
+        
+        ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
+        subscriptions.init(sessionsRepository);
+        SessionRegistry sessions = new SessionRegistry(subscriptions);
+        final PostOffice dispatcher = new PostOffice(subscriptions, authorizatorPolicy, new MemoryRetainedRepository(), sessions);
+        final BrokerConfiguration brokerConfig = new BrokerConfiguration(config);
+        MQTTConnectionFactory connectionFactory = new MQTTConnectionFactory(brokerConfig, authenticator, sessions,
+                                                                            dispatcher);
+
+        final NewNettyMQTTHandler mqttHandler = new NewNettyMQTTHandler(connectionFactory);
+        acceptor = new NewNettyAcceptor();
+        acceptor.initialize(mqttHandler, config, sslCtxCreator);
+
+        final long startTime = System.currentTimeMillis() - start;
+        LOG.info("Moquette server has been started successfully in {} ms", startTime);
+        initialized = true;
+    }
+
+    public void stopServer() {
+        LOG.info("Unbinding server from the configured ports");
+        acceptor.close();
+        LOG.trace("Stopping MQTT protocol processor");
+//        m_processorBootstrapper.shutdown();
+        initialized = false;
+
+        // calling shutdown() does not actually stop tasks that are not cancelled,
+        // and SessionsRepository does not stop its tasks. Thus shutdownNow().
+        scheduler.shutdownNow();
+
+        LOG.info("Moquette server has been stopped.");
+    }
+
+    private static File defaultConfigFile() {
+        String configPath = System.getProperty("moquette.path", null);
+        return new File(configPath, IConfig.DEFAULT_CONFIG);
+    }
+}

broker/src/main/java/io/moquette/broker/Session.java

@@ -0,0 +1,341 @@
+package io.moquette.broker;
+
+import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.netty.buffer.ByteBuf;
+import io.netty.handler.codec.mqtt.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+class Session {
+
+    private static final Logger LOG = LoggerFactory.getLogger(Session.class);
+    private static final int FLIGHT_BEFORE_RESEND_MS = 5_000;
+    private static final int INFLIGHT_WINDOW_SIZE = 10;
+
+    class InFlightPacket implements Delayed {
+
+        final int packetId;
+        private long startTime;
+
+        InFlightPacket(int packetId, long delayInMilliseconds) {
+            this.packetId = packetId;
+            this.startTime = System.currentTimeMillis() + delayInMilliseconds;
+        }
+
+        @Override
+        public long getDelay(TimeUnit unit) {
+            long diff = startTime - System.currentTimeMillis();
+            return unit.convert(diff, TimeUnit.MILLISECONDS);
+        }
+
+        @Override
+        public int compareTo(Delayed o) {
+            if ((this.startTime - ((InFlightPacket) o).startTime) == 0) {
+                return 0;
+            }
+            if ((this.startTime - ((InFlightPacket) o).startTime) > 0) {
+                return 1;
+            } else {
+                return -1;
+            }
+        }
+    }
+
+    enum SessionStatus {
+        CONNECTED, CONNECTING, DISCONNECTING, DISCONNECTED
+    }
+
+    static final class Will {
+
+        final String topic;
+        final ByteBuf payload;
+        final MqttQoS qos;
+        final boolean retained;
+
+        Will(String topic, ByteBuf payload, MqttQoS qos, boolean retained) {
+            this.topic = topic;
+            this.payload = payload;
+            this.qos = qos;
+            this.retained = retained;
+        }
+    }
+
+    private final String clientId;
+    private boolean clean;
+    private Will will;
+    private Queue<SessionRegistry.EnqueuedMessage> sessionQueue;
+    private final AtomicReference<SessionStatus> status = new AtomicReference<>(SessionStatus.DISCONNECTED);
+    private MQTTConnection mqttConnection;
+    private List<Subscription> subscriptions = new ArrayList<>();
+    private final Map<Integer, SessionRegistry.EnqueuedMessage> inflightWindow = new HashMap<>();
+    private final DelayQueue<InFlightPacket> inflightTimeouts = new DelayQueue<>();
+    private final Map<Integer, MqttPublishMessage> qos2Receiving = new HashMap<>();
+    private final AtomicInteger inflightSlots = new AtomicInteger(INFLIGHT_WINDOW_SIZE); // this should be configurable
+
+    Session(String clientId, boolean clean, Will will, Queue<SessionRegistry.EnqueuedMessage> sessionQueue) {
+        this.clientId = clientId;
+        this.clean = clean;
+        this.will = will;
+        this.sessionQueue = sessionQueue;
+    }
+
+    Session(boolean clean, String clientId, Queue<SessionRegistry.EnqueuedMessage> sessionQueue) {
+        this.clientId = clientId;
+        this.clean = clean;
+        this.sessionQueue = sessionQueue;
+    }
+
+    void update(boolean clean, Will will) {
+        this.clean = clean;
+        this.will = will;
+    }
+
+    void markConnected() {
+        assignState(SessionStatus.DISCONNECTED, SessionStatus.CONNECTED);
+    }
+
+    void bind(MQTTConnection mqttConnection) {
+        this.mqttConnection = mqttConnection;
+    }
+
+    public boolean disconnected() {
+        return status.get() == SessionStatus.DISCONNECTED;
+    }
+
+    public boolean connected() {
+        return status.get() == SessionStatus.CONNECTED;
+    }
+
+    public String getClientID() {
+        return clientId;
+    }
+
+    public List<Subscription> getSubscriptions() {
+        return new ArrayList<>(subscriptions);
+    }
+
+    public void addSubscriptions(List<Subscription> newSubscriptions) {
+        subscriptions.addAll(newSubscriptions);
+    }
+
+    public boolean hasWill() {
+        return will != null;
+    }
+
+    public Will getWill() {
+        return will;
+    }
+
+    boolean assignState(SessionStatus expected, SessionStatus newState) {
+        return status.compareAndSet(expected, newState);
+    }
+
+    public void closeImmediately() {
+        mqttConnection.dropConnection();
+    }
+
+    public void disconnect() {
+        final boolean res = assignState(SessionStatus.CONNECTED, SessionStatus.DISCONNECTING);
+        if (!res) {
+            // someone already moved away from CONNECTED
+            // TODO what to do?
+            return;
+        }
+
+        mqttConnection = null;
+        will = null;
+
+        assignState(SessionStatus.DISCONNECTING, SessionStatus.DISCONNECTED);
+    }
+
+    boolean isClean() {
+        return clean;
+    }
+
+    public void processPubRec(int packetId) {
+        inflightWindow.remove(packetId);
+        inflightSlots.incrementAndGet();
+        if (canSkipQueue()) {
+            inflightSlots.decrementAndGet();
+            int pubRelPacketId = mqttConnection.nextPacketId();
+            inflightWindow.put(pubRelPacketId, new SessionRegistry.PubRelMarker());
+            inflightTimeouts.add(new InFlightPacket(pubRelPacketId, FLIGHT_BEFORE_RESEND_MS));
+            MqttMessage pubRel = MQTTConnection.pubrel(pubRelPacketId);
+            mqttConnection.sendIfWritableElseDrop(pubRel);
+
+            drainQueueToConnection();
+        } else {
+            sessionQueue.add(new SessionRegistry.PubRelMarker());
+        }
+    }
+
+    public void processPubComp(int messageID) {
+        inflightWindow.remove(messageID);
+        inflightSlots.incrementAndGet();
+
+        drainQueueToConnection();
+
+        // TODO notify the interceptor
+//                final InterceptAcknowledgedMessage interceptAckMsg = new InterceptAcknowledgedMessage(inflightMsg, topic,
+//                    username, messageID);
+//                m_interceptor.notifyMessageAcknowledged(interceptAckMsg);
+    }
+
+    public void sendPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload) {
+        switch (qos) {
+            case AT_MOST_ONCE:
+                if (connected()) {
+                    mqttConnection.sendPublishNotRetainedQos0(topic, qos, payload);
+                }
+                break;
+            case AT_LEAST_ONCE:
+                sendPublishQos1(topic, qos, payload);
+                break;
+            case EXACTLY_ONCE:
+                sendPublishQos2(topic, qos, payload);
+                break;
+            case FAILURE:
+                LOG.error("Not admissible");
+        }
+    }
+
+    private void sendPublishQos1(Topic topic, MqttQoS qos, ByteBuf payload) {
+        if (!connected() && isClean()) {
+            //pushing messages to disconnected not clean session
+            return;
+        }
+
+        if (canSkipQueue()) {
+            inflightSlots.decrementAndGet();
+            int packetId = mqttConnection.nextPacketId();
+            inflightWindow.put(packetId, new SessionRegistry.PublishedMessage(topic, qos, payload));
+            inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
+            MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos, payload, packetId);
+            mqttConnection.sendPublish(publishMsg);
+
+            // TODO drainQueueToConnection();?
+        } else {
+            final SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(topic, qos, payload);
+            sessionQueue.add(msg);
+        }
+    }
+
+    private void sendPublishQos2(Topic topic, MqttQoS qos, ByteBuf payload) {
+        if (canSkipQueue()) {
+            inflightSlots.decrementAndGet();
+            int packetId = mqttConnection.nextPacketId();
+            inflightWindow.put(packetId, new SessionRegistry.PublishedMessage(topic, qos, payload));
+            inflightTimeouts.add(new InFlightPacket(packetId, FLIGHT_BEFORE_RESEND_MS));
+            MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(topic.toString(), qos, payload, packetId);
+            mqttConnection.sendPublish(publishMsg);
+
+            drainQueueToConnection();
+        } else {
+            final SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(topic, qos, payload);
+            sessionQueue.add(msg);
+        }
+    }
+
+    private boolean canSkipQueue() {
+        return sessionQueue.isEmpty() &&
+            inflightSlots.get() > 0 &&
+            connected() &&
+            mqttConnection.channel.isWritable();
+    }
+
+    void pubAckReceived(int ackPacketId) {
+        // TODO remain to invoke in somehow m_interceptor.notifyMessageAcknowledged
+        inflightWindow.remove(ackPacketId);
+        inflightSlots.incrementAndGet();
+        drainQueueToConnection();
+    }
+
+    public void resendInflightNotAcked() {
+        Collection<InFlightPacket> expired = new ArrayList<>(INFLIGHT_WINDOW_SIZE);
+        inflightTimeouts.drainTo(expired);
+
+        for (InFlightPacket notAckPacketId : expired) {
+            if (inflightWindow.containsKey(notAckPacketId.packetId)) {
+                final SessionRegistry.PublishedMessage msg = (SessionRegistry.PublishedMessage) inflightWindow.get(notAckPacketId.packetId);
+                final Topic topic = msg.topic;
+                final MqttQoS qos = msg.publishingQos;
+                final ByteBuf payload = msg.payload;
+
+                MqttPublishMessage publishMsg = publishNotRetainedDuplicated(notAckPacketId, topic, qos, payload);
+                mqttConnection.sendPublish(publishMsg);
+            }
+        }
+    }
+
+    private MqttPublishMessage publishNotRetainedDuplicated(InFlightPacket notAckPacketId, Topic topic, MqttQoS qos, ByteBuf payload) {
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, true, qos, false, 0);
+        MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic.toString(), notAckPacketId.packetId);
+        return new MqttPublishMessage(fixedHeader, varHeader, payload);
+    }
+
+    private void drainQueueToConnection() {
+        // consume the queue
+        while (!canSkipQueue()) {
+            final SessionRegistry.EnqueuedMessage msg = sessionQueue.remove();
+            inflightSlots.decrementAndGet();
+            int sendPacketId = mqttConnection.nextPacketId();
+            inflightWindow.put(sendPacketId, msg);
+            if (msg instanceof SessionRegistry.PubRelMarker) {
+                MqttMessage pubRel = MQTTConnection.pubrel(sendPacketId);
+                mqttConnection.sendIfWritableElseDrop(pubRel);
+            } else {
+                final SessionRegistry.PublishedMessage msgPub = (SessionRegistry.PublishedMessage) msg;
+                MqttPublishMessage publishMsg = MQTTConnection.notRetainedPublishWithMessageId(msgPub.topic.toString(),
+                    msgPub.publishingQos,
+                    msgPub.payload, sendPacketId);
+                mqttConnection.sendPublish(publishMsg);
+            }
+        }
+    }
+
+    public void writabilityChanged() {
+        drainQueueToConnection();
+    }
+
+    public void sendQueuedMessagesWhileOffline() {
+        LOG.trace("Republishing all saved messages for session {} on CId={}", this, this.clientId);
+        drainQueueToConnection();
+    }
+
+    void sendRetainedPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload) {
+        if (qos != MqttQoS.AT_MOST_ONCE) {
+            // QoS 1 or 2
+            mqttConnection.sendPublishRetainedWithPacketId(topic, qos, payload);
+        } else {
+            mqttConnection.sendPublishRetainedQos0(topic, qos, payload);
+        }
+    }
+
+    public void receivedPublishQos2(int messageID, MqttPublishMessage msg) {
+        qos2Receiving.put(messageID, msg);
+        msg.retain();
+        mqttConnection.sendPublishReceived(messageID);
+    }
+
+    public void receivedPubRelQos2(int messageID) {
+        qos2Receiving.remove(messageID);
+    }
+
+    @Override
+    public String toString() {
+        return "Session{" +
+            "clientId='" + clientId + '\'' +
+            ", clean=" + clean +
+            ", status=" + status +
+            ", inflightSlots=" + inflightSlots +
+            '}';
+    }
+}

broker/src/main/java/io/moquette/broker/SessionCorruptedException.java

@@ -0,0 +1,10 @@
+package io.moquette.broker;
+
+public class SessionCorruptedException extends RuntimeException {
+
+    private static final long serialVersionUID = 5848069213104389412L;
+
+    SessionCorruptedException(String msg) {
+        super(msg);
+    }
+}

broker/src/main/java/io/moquette/broker/SessionRegistry.java

@@ -0,0 +1,220 @@
+package io.moquette.broker;
+
+import io.moquette.broker.Session.SessionStatus;
+import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
+import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Topic;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.handler.codec.mqtt.MqttConnectMessage;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+
+class SessionRegistry {
+
+    static abstract class EnqueuedMessage {
+    }
+
+    static class PublishedMessage extends EnqueuedMessage {
+
+        final Topic topic;
+        final MqttQoS publishingQos;
+        final ByteBuf payload;
+
+        PublishedMessage(Topic topic, MqttQoS publishingQos, ByteBuf payload) {
+            this.topic = topic;
+            this.publishingQos = publishingQos;
+            this.payload = payload;
+        }
+    }
+
+    static final class PubRelMarker extends EnqueuedMessage {
+    }
+
+    private enum PostConnectAction {
+        NONE, SEND_STORED_MESSAGES
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(SessionRegistry.class);
+
+    private final ConcurrentMap<String, Session> pool = new ConcurrentHashMap<>();
+    private final ISubscriptionsDirectory subscriptionsDirectory;
+    private final ConcurrentMap<String, Queue<SessionRegistry.EnqueuedMessage>> queues = new ConcurrentHashMap<>();
+
+    SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory) {
+        this.subscriptionsDirectory = subscriptionsDirectory;
+    }
+
+    void bindToSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) {
+        boolean isSessionAlreadyStored = false;
+        PostConnectAction postConnectAction = PostConnectAction.NONE;
+        if (!pool.containsKey(clientId)) {
+            // case 1
+            final Session newSession = createNewSession(mqttConnection, msg, clientId);
+
+            // publish the session
+            final Session previous = pool.putIfAbsent(clientId, newSession);
+            final boolean success = previous == null;
+
+            if (success) {
+                LOG.trace("case 1, not existing session with CId {}", clientId);
+            } else {
+                postConnectAction = bindToExistingSession(mqttConnection, msg, clientId, newSession);
+                isSessionAlreadyStored = true;
+            }
+        } else {
+            final Session newSession = createNewSession(mqttConnection, msg, clientId);
+            postConnectAction = bindToExistingSession(mqttConnection, msg, clientId, newSession);
+            isSessionAlreadyStored = true;
+        }
+        final boolean msgCleanSessionFlag = msg.variableHeader().isCleanSession();
+        boolean isSessionAlreadyPresent = (!msgCleanSessionFlag && isSessionAlreadyStored);
+        mqttConnection.sendConnAck(isSessionAlreadyPresent);
+
+        if (postConnectAction == PostConnectAction.SEND_STORED_MESSAGES) {
+            final Session session = pool.get(clientId);
+            session.sendQueuedMessagesWhileOffline();
+        }
+    }
+
+    private PostConnectAction bindToExistingSession(MQTTConnection mqttConnection, MqttConnectMessage msg,
+                                                    String clientId, Session newSession) {
+        PostConnectAction postConnectAction = PostConnectAction.NONE;
+        final boolean newIsClean = msg.variableHeader().isCleanSession();
+        final Session oldSession = pool.get(clientId);
+        if (newIsClean && oldSession.disconnected()) {
+            // case 2
+            dropQueuesForClient(clientId);
+            unsubscribe(oldSession);
+
+            // publish new session
+            boolean result = oldSession.assignState(SessionStatus.DISCONNECTED, SessionStatus.CONNECTING);
+            if (!result) {
+                throw new SessionCorruptedException("old session was already changed state");
+            }
+            copySessionConfig(msg, oldSession);
+            oldSession.bind(mqttConnection);
+
+            result = oldSession.assignState(SessionStatus.CONNECTING, SessionStatus.CONNECTED);
+            if (!result) {
+                throw new SessionCorruptedException("old session moved in connected state by other thread");
+            }
+            final boolean published = pool.replace(clientId, oldSession, oldSession);
+            if (!published) {
+                throw new SessionCorruptedException("old session was already removed");
+            }
+            LOG.trace("case 2, oldSession with same CId {} disconnected", clientId);
+        } else if (!newIsClean && oldSession.disconnected()) {
+            // case 3
+            reactivateSubscriptions(oldSession);
+
+            // mark as connected
+            final boolean connecting = oldSession.assignState(SessionStatus.DISCONNECTED, SessionStatus.CONNECTING);
+            if (!connecting) {
+                throw new SessionCorruptedException("old session moved in connected state by other thread");
+            }
+            oldSession.bind(mqttConnection);
+
+            final boolean connected = oldSession.assignState(SessionStatus.CONNECTING, SessionStatus.CONNECTED);
+            if (!connected) {
+                throw new SessionCorruptedException("old session moved in other state state by other thread");
+            }
+
+            // publish new session
+            final boolean published = pool.replace(clientId, oldSession, oldSession);
+            if (!published) {
+                throw new SessionCorruptedException("old session was already removed");
+            }
+            postConnectAction = PostConnectAction.SEND_STORED_MESSAGES;
+            LOG.trace("case 3, oldSession with same CId {} disconnected", clientId);
+        } else if (oldSession.connected()) {
+            // case 4
+            LOG.trace("case 4, oldSession with same CId {} still connected, force to close", clientId);
+            oldSession.closeImmediately();
+            //remove(clientId);
+            // publish new session
+            final boolean published = pool.replace(clientId, oldSession, newSession);
+            if (!published) {
+                throw new SessionCorruptedException("old session was already removed");
+            }
+        }
+        // case not covered new session is clean true/false and old session not in CONNECTED/DISCONNECTED
+        return postConnectAction;
+    }
+
+    private void reactivateSubscriptions(Session session) {
+        for (Subscription existingSub : session.getSubscriptions()) {
+            // TODO
+//            subscriptionsDirectory.reactivate(existingSub.getTopicFilter(), session.getClientID());
+        }
+    }
+
+    private void unsubscribe(Session session) {
+        for (Subscription existingSub : session.getSubscriptions()) {
+            subscriptionsDirectory.removeSubscription(existingSub.getTopicFilter(), session.getClientID());
+        }
+    }
+
+    private Session createNewSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) {
+        final Queue<SessionRegistry.EnqueuedMessage> sessionQueue = queues.computeIfAbsent(clientId, (String cli) -> new ConcurrentLinkedQueue<>());
+        final boolean clean = msg.variableHeader().isCleanSession();
+        final Session newSession;
+        if (msg.variableHeader().isWillFlag()) {
+            final Session.Will will = createWill(msg);
+            newSession = new Session(clientId, clean, will, sessionQueue);
+        } else {
+            newSession = new Session(clean, clientId, sessionQueue);
+        }
+
+        newSession.markConnected();
+        newSession.bind(mqttConnection);
+
+        return newSession;
+    }
+
+    private void copySessionConfig(MqttConnectMessage msg, Session session) {
+        final boolean clean = msg.variableHeader().isCleanSession();
+        final Session.Will will;
+        if (msg.variableHeader().isWillFlag()) {
+            will = createWill(msg);
+        } else {
+            will = null;
+        }
+        session.update(clean, will);
+    }
+
+    private Session.Will createWill(MqttConnectMessage msg) {
+        final ByteBuf willPayload = Unpooled.copiedBuffer(msg.payload().willMessageInBytes());
+        final String willTopic = msg.payload().willTopic();
+        final boolean retained = msg.variableHeader().isWillRetain();
+        final MqttQoS qos = MqttQoS.valueOf(msg.variableHeader().willQos());
+        return new Session.Will(willTopic, willPayload, qos, retained);
+    }
+
+    Session retrieve(String clientID) {
+        return pool.get(clientID);
+    }
+
+    public void remove(String clientID) {
+        pool.remove(clientID);
+    }
+
+    public void disconnect(String clientID) {
+        final Session session = retrieve(clientID);
+        if (session == null) {
+            LOG.debug("Some other thread already removed the session CId={}", clientID);
+            return;
+        }
+        session.disconnect();
+    }
+
+    private void dropQueuesForClient(String clientId) {
+        queues.remove(clientId);
+    }
+}

broker/src/main/java/io/moquette/server/Server.java

@@ -25,7 +25,7 @@ import io.moquette.spi.impl.ProtocolProcessor;
 import io.moquette.spi.impl.ProtocolProcessorBootstrapper;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.security.IAuthenticator;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.moquette.spi.security.ISslContextCreator;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
 import org.slf4j.Logger;
@@ -141,7 +141,7 @@ public class Server {
     }
 
     public void startServer(IConfig config, List<? extends InterceptHandler> handlers, ISslContextCreator sslCtxCreator,
-                            IAuthenticator authenticator, IAuthorizator authorizator) throws IOException {
+                            IAuthenticator authenticator, IAuthorizatorPolicy authorizator) throws IOException {
         final long start = System.currentTimeMillis();
         if (handlers == null) {
             handlers = Collections.emptyList();

broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java

@@ -27,7 +27,7 @@ import io.netty.handler.timeout.IdleStateEvent;
 import static io.netty.channel.ChannelFutureListener.CLOSE_ON_FAILURE;
 
 @Sharable
-class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler {
+public class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler {
 
     private static final Logger LOG = LoggerFactory.getLogger(MoquetteIdleTimeoutHandler.class);
 

broker/src/main/java/io/moquette/spi/impl/ConnectHandler.java

@@ -10,7 +10,7 @@ import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.impl.subscriptions.Topic;
 import io.moquette.spi.security.IAuthenticator;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
@@ -40,7 +40,7 @@ public class ConnectHandler {
     private SessionsRepository sessionsRepository;
     private ISessionsStore m_sessionsStore;
     private IAuthenticator m_authenticator;
-    private IAuthorizator m_authorizator;
+    private IAuthorizatorPolicy m_authorizator;
     private ISubscriptionsDirectory subscriptions;
     private boolean allowAnonymous;
     private boolean allowZeroByteClientId;
@@ -48,7 +48,7 @@ public class ConnectHandler {
     private InternalRepublisher internalRepublisher;
 
     ConnectHandler(IConnectionsManager connectedClients, BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
-                   ISessionsStore sessionsStore, IAuthenticator authenticator, IAuthorizator authorizator,
+                   ISessionsStore sessionsStore, IAuthenticator authenticator, IAuthorizatorPolicy authorizator,
                    ISubscriptionsDirectory subscriptions, boolean allowAnonymous, boolean allowZeroByteClientId,
                    boolean reauthorizeSubscriptionsOnConnect, InternalRepublisher internalRepublisher) {
         this.connectionDescriptors = connectedClients;

broker/src/main/java/io/moquette/spi/impl/DebugUtils.java

@@ -20,10 +20,19 @@ import io.netty.buffer.ByteBuf;
 
 import java.nio.charset.StandardCharsets;
 
-final class DebugUtils {
+public final class DebugUtils {
 
-    static String payload2Str(ByteBuf content) {
-        return new String(content.copy().array(), StandardCharsets.UTF_8);
+    public static String payload2Str(ByteBuf content) {
+        final ByteBuf copy = content.copy();
+        final byte[] bytesContent;
+        if (copy.isDirect()) {
+            final int size = copy.readableBytes();
+            bytesContent = new byte[size];
+            copy.readBytes(bytesContent);
+        } else {
+            bytesContent = copy.array();
+        }
+        return new String(bytesContent, StandardCharsets.UTF_8);
     }
 
     private DebugUtils() {

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -31,7 +31,7 @@ import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.impl.subscriptions.Topic;
 import io.moquette.spi.security.IAuthenticator;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.handler.codec.mqtt.*;
@@ -41,11 +41,9 @@ import org.slf4j.LoggerFactory;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
-import static io.moquette.server.ConnectionDescriptor.ConnectionState.*;
 import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
 import static io.moquette.spi.impl.Utils.messageId;
 import static io.moquette.spi.impl.Utils.readBytesAndRewind;
@@ -72,7 +70,7 @@ public class ProtocolProcessor {
     private ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
 
     private ISubscriptionsDirectory subscriptions;
-    private IAuthorizator m_authorizator;
+    private IAuthorizatorPolicy m_authorizator;
 
     private IMessagesStore m_messagesStore;
 
@@ -94,7 +92,7 @@ public class ProtocolProcessor {
     }
 
     public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
-                     IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
+                     IAuthenticator authenticator, boolean allowAnonymous, IAuthorizatorPolicy authorizator,
                      BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
                      boolean reauthorizeSubscriptionsOnConnect) {
         init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false,
@@ -103,7 +101,7 @@ public class ProtocolProcessor {
 
     public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                      IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
-                     IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
+                     IAuthorizatorPolicy authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
                      boolean reauthorizeSubscriptionsOnConnect) {
         init(new ConnectionDescriptorStore(), subscriptions, storageService, sessionsStore,
              authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository,
@@ -130,7 +128,7 @@ public class ProtocolProcessor {
      */
     void init(IConnectionsManager connectionDescriptors, ISubscriptionsDirectory subscriptions,
               IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
-              boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
+              boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizatorPolicy authorizator,
               BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
               boolean reauthorizeSubscriptionsOnConnect) {
         LOG.info("Initializing MQTT protocol processor..");

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java

@@ -31,7 +31,7 @@ import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
 import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.security.IAuthenticator;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -82,7 +82,7 @@ public class ProtocolProcessorBootstrapper {
      * @return the processor created for the broker.
      */
     public ProtocolProcessor init(IConfig props, List<? extends InterceptHandler> embeddedObservers,
-            IAuthenticator authenticator, IAuthorizator authorizator, Server server) {
+                                  IAuthenticator authenticator, IAuthorizatorPolicy authorizator, Server server) {
         IMessagesStore messagesStore;
         LOG.info("Initializing messages and sessions stores...");
         String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, INMEMDB_STORE_CLASS);
@@ -145,13 +145,13 @@ public class ProtocolProcessorBootstrapper {
         LOG.info("Configuring MQTT authorizator...");
         String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "");
         if (authorizator == null && !authorizatorClassName.isEmpty()) {
-            authorizator = loadClass(authorizatorClassName, IAuthorizator.class, IConfig.class, props);
+            authorizator = loadClass(authorizatorClassName, IAuthorizatorPolicy.class, IConfig.class, props);
         }
 
         if (authorizator == null) {
             String aclFilePath = props.getProperty(BrokerConstants.ACL_FILE_PROPERTY_NAME, "");
             if (aclFilePath != null && !aclFilePath.isEmpty()) {
-                authorizator = new DenyAllAuthorizator();
+                authorizator = new DenyAllAuthorizatorPolicy();
                 try {
                     LOG.info("Parsing ACL file. Path = {}", aclFilePath);
                     authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath));
@@ -159,7 +159,7 @@ public class ProtocolProcessorBootstrapper {
                     LOG.error("Unable to parse ACL file. path=" + aclFilePath, pex);
                 }
             } else {
-                authorizator = new PermitAllAuthorizator();
+                authorizator = new PermitAllAuthorizatorPolicy();
             }
             LOG.info("An {} authorizator instance will be used", authorizator.getClass().getName());
         }

broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java

@@ -19,7 +19,7 @@ package io.moquette.spi.impl;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.netty.channel.Channel;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
 import org.slf4j.Logger;
@@ -34,7 +34,7 @@ class Qos0PublishHandler extends QosPublishHandler {
     private final BrokerInterceptor m_interceptor;
     private final MessagesPublisher publisher;
 
-    public Qos0PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
+    public Qos0PublishHandler(IAuthorizatorPolicy authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
                               MessagesPublisher messagesPublisher) {
         super(authorizator);
         this.m_messagesStore = messagesStore;

broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java

@@ -18,11 +18,10 @@ package io.moquette.spi.impl;
 
 import io.moquette.connections.IConnectionsManager;
 import io.moquette.server.ConnectionDescriptor;
-import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.netty.channel.Channel;
 import io.netty.handler.codec.mqtt.MqttFixedHeader;
 import io.netty.handler.codec.mqtt.MqttMessageType;
@@ -47,7 +46,7 @@ class Qos1PublishHandler extends QosPublishHandler {
     private final IConnectionsManager connectionDescriptors;
     private final MessagesPublisher publisher;
 
-    Qos1PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
+    Qos1PublishHandler(IAuthorizatorPolicy authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor,
                        IConnectionsManager connectionDescriptors, MessagesPublisher messagesPublisher) {
         super(authorizator);
         this.m_messagesStore = messagesStore;

broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java

@@ -18,14 +18,12 @@ package io.moquette.spi.impl;
 
 import io.moquette.connections.IConnectionsManager;
 import io.moquette.server.ConnectionDescriptor;
-import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.ClientSession;
 import io.moquette.spi.IMessagesStore;
-import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.netty.channel.Channel;
 import io.netty.handler.codec.mqtt.*;
 import org.slf4j.Logger;
@@ -50,7 +48,7 @@ class Qos2PublishHandler extends QosPublishHandler {
     private final MessagesPublisher publisher;
     private final SessionsRepository sessionsRepository;
 
-    Qos2PublishHandler(IAuthorizator authorizator, ISubscriptionsDirectory subscriptions,
+    Qos2PublishHandler(IAuthorizatorPolicy authorizator, ISubscriptionsDirectory subscriptions,
                        IMessagesStore messagesStore, BrokerInterceptor interceptor,
                        IConnectionsManager connectionDescriptors,
                        MessagesPublisher messagesPublisher, SessionsRepository sessionsRepository) {

broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java

@@ -20,16 +20,16 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import io.netty.channel.Channel;
 
 abstract class QosPublishHandler {
 
     private static final Logger LOG = LoggerFactory.getLogger(QosPublishHandler.class);
 
-    protected final IAuthorizator m_authorizator;
+    protected final IAuthorizatorPolicy m_authorizator;
 
-    protected QosPublishHandler(IAuthorizator m_authorizator) {
+    protected QosPublishHandler(IAuthorizatorPolicy m_authorizator) {
         this.m_authorizator = m_authorizator;
     }
 

broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java

@@ -17,7 +17,7 @@
 package io.moquette.spi.impl.security;
 
 import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 import java.text.ParseException;
 import java.util.*;
 
@@ -27,7 +27,7 @@ import java.util.*;
  *
  * Not thread safe.
  */
-class AuthorizationsCollector implements IAuthorizator {
+class AuthorizationsCollector implements IAuthorizatorPolicy {
 
     private List<Authorization> m_globalAuthorizations = new ArrayList<>();
     private List<Authorization> m_patternAuthorizations = new ArrayList<>();

broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizatorPolicy.java

@@ -17,9 +17,9 @@
 package io.moquette.spi.impl.security;
 
 import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 
-public class DenyAllAuthorizator implements IAuthorizator {
+public class DenyAllAuthorizatorPolicy implements IAuthorizatorPolicy {
 
     @Override
     public boolean canWrite(Topic topic, String user, String client) {

broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizatorPolicy.java

@@ -17,9 +17,9 @@
 package io.moquette.spi.impl.security;
 
 import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.security.IAuthorizator;
+import io.moquette.spi.security.IAuthorizatorPolicy;
 
-public class PermitAllAuthorizator implements IAuthorizator {
+public class PermitAllAuthorizatorPolicy implements IAuthorizatorPolicy {
 
     @Override
     public boolean canWrite(Topic topic, String user, String client) {

broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java

@@ -15,6 +15,8 @@
  */
 package io.moquette.spi.impl.subscriptions;
 
+import io.netty.handler.codec.mqtt.MqttQoS;
+
 import java.util.*;
 
 class CNode {
@@ -79,8 +81,8 @@ class CNode {
         this.children.remove(node);
     }
 
-    CNode addSubscription(String clientId, Topic topic) {
-        this.subscriptions.add(new Subscription(clientId, topic));
+    CNode addSubscription(Subscription newSubscription) {
+        this.subscriptions.add(new Subscription(newSubscription));
         return this;
     }
 

broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java

@@ -98,9 +98,45 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
      *            to use fo searching matching subscriptions.
      * @return the list of matching subscriptions, or empty if not matching.
      */
-     Set<Subscription> match(Topic topic) {
-        final Set<Subscription> matchingSubs = recursiveMatch(topic, this.root);
+    @Override
+    public Set<Subscription> matchWithoutQosSharpening(Topic topic) {
+        return recursiveMatch(topic, this.root);
+    }
+
+    @Override
+    public Set<Subscription> matchQosSharpening(Topic topic) {
+        final Set<Subscription> subscriptions = matchWithoutQosSharpening(topic);
+
+        Map<String, Subscription> subsGroupedByClient = new HashMap<>();
+        for (Subscription sub : subscriptions) {
+            Subscription existingSub = subsGroupedByClient.get(sub.clientId);
+            // update the selected subscriptions if not present or if has a greater qos
+            if (existingSub == null || existingSub.qosLessThan(sub)) {
+                subsGroupedByClient.put(sub.clientId, sub);
+            }
+        }
+        return new HashSet<>(subsGroupedByClient.values());
+    }
+
+    /**
+     * Given a topic string return the clients subscriptions that matches it. Topic string can't
+     * contain character # and + because they are reserved to listeners subscriptions, and not topic
+     * publishing.
+     *
+     * @param topic
+     *            to use fo searching matching subscriptions.
+     * @return the list of matching subscriptions, or empty if not matching.
+     */
+    @Deprecated
+    Set<Subscription> match(Topic topic) {
+        final Set<Subscription> matchingSubs = matchWithoutQosSharpening(topic);
         // remove the overlapping subscriptions, selecting ones with greatest qos
+        Map<String, Subscription> subsForClient = loadSubscriptionsFromStorage(matchingSubs);
+        return new HashSet<>(subsForClient.values());
+    }
+
+    @Deprecated
+    private Map<String, Subscription> loadSubscriptionsFromStorage(Set<Subscription> matchingSubs) {
         Map<String, Subscription> subsForClient = new HashMap<>();
         for (Subscription matchingSub : matchingSubs) {
             Subscription existingSub = subsForClient.get(matchingSub.clientId);
@@ -120,7 +156,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
                 subsForClient.put(sub.clientId, sub);
             }
         }
-        return new HashSet<>(subsForClient.values());
+        return subsForClient;
     }
 
     Set<Subscription> recursiveMatch(Topic topic, INode inode) {
@@ -171,28 +207,28 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
     public void add(Subscription newSubscription) {
         Action res;
         do {
-            res = insert(newSubscription.clientId, newSubscription.topicFilter, this.root, newSubscription.topicFilter);
+            res = insert(newSubscription.topicFilter, this.root, newSubscription);
         } while (res == Action.REPEAT);
     }
 
-    private Action insert(String clientId, Topic topic, final INode inode, Topic fullpath) {
+    private Action insert(Topic topic, final INode inode, Subscription newSubscription) {
         Token token = topic.headToken();
         if (!topic.isEmpty() && inode.mainNode().anyChildrenMatch(token)) {
             Topic remainingTopic = topic.exceptHeadToken();
             INode nextInode = inode.mainNode().childOf(token);
-            return insert(clientId, remainingTopic, nextInode, fullpath);
+            return insert(remainingTopic, nextInode, newSubscription);
         } else {
             if (topic.isEmpty()) {
-                return insertSubscription(clientId, fullpath, inode);
+                return insertSubscription(inode, newSubscription);
             } else {
-                return createNodeAndInsertSubscription(clientId, topic, inode, fullpath);
+                return createNodeAndInsertSubscription(topic, inode, newSubscription);
             }
         }
     }
 
-    private Action insertSubscription(String clientId, Topic topic, INode inode) {
+    private Action insertSubscription(INode inode, Subscription newSubscription) {
         CNode cnode = inode.mainNode();
-        CNode updatedCnode = cnode.copy().addSubscription(clientId, topic);
+        CNode updatedCnode = cnode.copy().addSubscription(newSubscription);
         if (inode.compareAndSet(cnode, updatedCnode)) {
             return Action.OK;
         } else {
@@ -200,8 +236,8 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
         }
     }
 
-    private Action createNodeAndInsertSubscription(String clientId, Topic topic, INode inode, Topic fullpath) {
-        INode newInode = createPathRec(clientId, topic, fullpath);
+    private Action createNodeAndInsertSubscription(Topic topic, INode inode, Subscription newSubscription) {
+        INode newInode = createPathRec(topic, newSubscription);
         CNode cnode = inode.mainNode();
         CNode updatedCnode = cnode.copy();
         updatedCnode.add(newInode);
@@ -209,24 +245,24 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
         return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT;
     }
 
-    private INode createLeafNodes(String clientId, Topic fullpath, Token token) {
+    private INode createLeafNodes(Token token, Subscription newSubscription) {
         CNode newLeafCnode = new CNode();
         newLeafCnode.token = token;
-        newLeafCnode.addSubscription(clientId, fullpath);
+        newLeafCnode.addSubscription(newSubscription);
 
         return new INode(newLeafCnode);
     }
 
-    private INode createPathRec(String clientId, Topic topic, Topic fullpath) {
+    private INode createPathRec(Topic topic, Subscription newSubscription) {
         Topic remainingTopic = topic.exceptHeadToken();
         if (!remainingTopic.isEmpty()) {
-            INode inode = createPathRec(clientId, remainingTopic, fullpath);
+            INode inode = createPathRec(remainingTopic, newSubscription);
             CNode cnode = new CNode();
             cnode.token = topic.headToken();
             cnode.add(inode);
             return new INode(cnode);
         } else {
-            return createLeafNodes(clientId, fullpath, topic.headToken());
+            return createLeafNodes(topic.headToken(), newSubscription);
         }
     }
 

broker/src/main/java/io/moquette/spi/impl/subscriptions/ISubscriptionsDirectory.java

@@ -18,11 +18,16 @@ package io.moquette.spi.impl.subscriptions;
 import io.moquette.spi.impl.SessionsRepository;
 
 import java.util.List;
+import java.util.Set;
 
 public interface ISubscriptionsDirectory {
 
     void init(SessionsRepository sessionsRepository);
 
+    Set<Subscription> matchWithoutQosSharpening(Topic topic);
+
+    Set<Subscription> matchQosSharpening(Topic topic);
+
     void add(Subscription newSubscription);
 
     void removeSubscription(Topic topic, String clientID);

broker/src/main/java/io/moquette/spi/impl/subscriptions/TNode.java

@@ -33,7 +33,7 @@ class TNode extends CNode {
     }
 
     @Override
-    CNode addSubscription(String clientId, Topic topic) {
+    CNode addSubscription(Subscription newSubscription) {
         throw new IllegalStateException("Can't be invoked on TNode");
     }
 

broker/src/main/java/io/moquette/spi/security/IAuthorizatorPolicy.java

@@ -24,7 +24,7 @@ import io.moquette.spi.impl.subscriptions.Topic;
  * Create an authorizator that matches topic names with same grammar of subscriptions. The # is
  * always a terminator and its the multilevel matcher. The + sign is the single level matcher.
  */
-public interface IAuthorizator {
+public interface IAuthorizatorPolicy {
 
     /**
      * Ask the implementation of the authorizator if the topic can be used in a publish.