Separated persistence layer, created submodule for MapDB persistence letting the broker use the in-memory message store

broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright (c) 2012-2017 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.persistence;
+
+import io.moquette.spi.IMessagesStore;
+import io.moquette.spi.IMatchingCondition;
+import io.moquette.spi.MessageGUID;
+import io.moquette.spi.impl.Utils;
+import io.moquette.spi.impl.subscriptions.Topic;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.*;
+import static io.moquette.spi.impl.Utils.defaultGet;
+
+/**
+ * @author andrea
+ */
+public class MemoryMessagesStore implements IMessagesStore {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class);
+
+    private Map<Topic, MessageGUID> m_retainedStore = new HashMap<>();
+    private Map<MessageGUID, StoredMessage> m_persistentMessageStore = new HashMap<>();
+    private Map<String, Map<Integer, MessageGUID>> m_messageToGuids = new HashMap<>();
+
+    MemoryMessagesStore() {
+    }
+
+    @Override
+    public void initStore() {
+    }
+
+    @Override
+    public void storeRetained(Topic topic, MessageGUID guid) {
+        m_retainedStore.put(topic, guid);
+    }
+
+    @Override
+    public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
+        LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
+
+        List<StoredMessage> results = new ArrayList<>();
+
+        for (Map.Entry<Topic, MessageGUID> entry : m_retainedStore.entrySet()) {
+            final MessageGUID guid = entry.getValue();
+            StoredMessage storedMsg = m_persistentMessageStore.get(guid);
+            if (condition.match(entry.getKey())) {
+                results.add(storedMsg);
+            }
+        }
+
+        return results;
+    }
+
+    @Override
+    public MessageGUID storePublishForFuture(StoredMessage storedMessage) {
+        LOG.debug("storePublishForFuture store evt {}", storedMessage);
+        MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
+        storedMessage.setGuid(guid);
+        m_persistentMessageStore.put(guid, storedMessage);
+        final HashMap<Integer, MessageGUID> emptyGuids = new HashMap<>();
+        Map<Integer, MessageGUID> guids = defaultGet(m_messageToGuids, storedMessage.getClientID(), emptyGuids);
+        guids.put(storedMessage.getMessageID(), guid);
+        m_messageToGuids.put(storedMessage.getClientID(), guids);
+        return guid;
+    }
+
+    @Override
+    public void dropMessagesInSession(String clientID) {
+        Map<Integer, MessageGUID> messageGUIDMap = m_messageToGuids.get(clientID);
+        if (messageGUIDMap == null || messageGUIDMap.isEmpty()) {
+            return;
+        }
+        //remove all guids from retained
+        Collection<MessageGUID> messagesToRemove = new HashSet<>(messageGUIDMap.values());
+        messagesToRemove.removeAll(m_retainedStore.values());
+
+        for (MessageGUID guid : messagesToRemove) {
+            m_persistentMessageStore.remove(guid);
+        }
+    }
+
+    @Override
+    public StoredMessage getMessageByGuid(MessageGUID guid) {
+        return m_persistentMessageStore.get(guid);
+    }
+
+    @Override
+    public void cleanRetained(Topic topic) {
+        m_retainedStore.remove(topic);
+    }
+
+    @Override
+    public int getPendingPublishMessages(String clientID) {
+        Map<Integer, MessageGUID> messageToGuids = m_messageToGuids.get(clientID);
+        if (messageToGuids == null)
+            return 0;
+        else
+            return messageToGuids.size();
+    }
+
+    @Override
+    public MessageGUID mapToGuid(String clientID, int messageID) {
+        final HashMap<Integer, MessageGUID> emptyGuids = new HashMap<>();
+        Map<Integer, MessageGUID> guids = Utils.defaultGet(m_messageToGuids, clientID, emptyGuids);
+        return guids.get(messageID);
+    }
+}

broker/src/main/java/io/moquette/persistence/MemorySessionStore.java

@@ -0,0 +1,268 @@
+/*
+ * Copyright (c) 2012-2017 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.persistence;
+
+import io.moquette.server.Constants;
+import io.moquette.spi.ClientSession;
+import io.moquette.spi.IMessagesStore;
+import io.moquette.spi.IMessagesStore.StoredMessage;
+import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.MessageGUID;
+import io.moquette.spi.impl.Utils;
+import io.moquette.spi.impl.subscriptions.Subscription;
+import io.moquette.spi.impl.subscriptions.Topic;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.*;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ *
+ * @author andrea
+ */
+public class MemorySessionStore implements ISessionsStore {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
+
+    private Map<String, Map<Topic, Subscription>> m_persistentSubscriptions = new HashMap<>();
+
+    private Map<String, PersistentSession> m_persistentSessions = new HashMap<>();
+
+    // maps clientID->[MessageId -> guid]
+    private Map<String, Map<Integer, MessageGUID>> m_inflightStore = new HashMap<>();
+    // private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
+    // maps clientID->BlockingQueue
+    private Map<String, BlockingQueue<StoredMessage>> queues = new HashMap<>();
+    // maps clientID->[MessageId -> guid]
+    private Map<String, Map<Integer, MessageGUID>> m_secondPhaseStore = new HashMap<>();
+
+    private final IMessagesStore m_messagesStore;
+
+    public MemorySessionStore(IMessagesStore messagesStore) {
+        this.m_messagesStore = messagesStore;
+    }
+
+    @Override
+    public void removeSubscription(Topic topic, String clientID) {
+        LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
+            return;
+        }
+        Map<Topic, Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
+        clientSubscriptions.remove(topic);
+    }
+
+    @Override
+    public void initStore() {
+
+    }
+
+    @Override
+    public void addNewSubscription(Subscription newSubscription) {
+        final String clientID = newSubscription.getClientId();
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
+            m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
+        }
+
+        m_persistentSubscriptions.get(clientID).put(newSubscription.getTopicFilter(), newSubscription);
+    }
+
+    @Override
+    public void wipeSubscriptions(String clientID) {
+        m_persistentSubscriptions.remove(clientID);
+    }
+
+    @Override
+    public boolean contains(String clientID) {
+        return m_persistentSubscriptions.containsKey(clientID);
+    }
+
+    @Override
+    public ClientSession createNewSession(String clientID, boolean cleanSession) {
+        LOG.debug("createNewSession for client <{}>", clientID);
+        if (m_persistentSessions.containsKey(clientID)) {
+            LOG.error("already exists a session for client <{}>, bad condition", clientID);
+            throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
+        }
+        LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
+        m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
+        m_persistentSessions.put(clientID, new PersistentSession(cleanSession));
+        return new ClientSession(clientID, m_messagesStore, this, cleanSession);
+    }
+
+    @Override
+    public ClientSession sessionForClient(String clientID) {
+        if (!m_persistentSessions.containsKey(clientID)) {
+            return null;
+        }
+
+        PersistentSession storedSession = m_persistentSessions.get(clientID);
+        return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
+    }
+
+    @Override
+    public Collection<ClientSession> getAllSessions() {
+        Collection<ClientSession> result = new ArrayList<>();
+        for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
+            result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
+        }
+        return result;
+    }
+
+    @Override
+    public void updateCleanStatus(String clientID, boolean cleanSession) {
+        m_persistentSessions.put(clientID, new PersistentSession(cleanSession));
+    }
+
+    @Override
+    public List<ClientTopicCouple> listAllSubscriptions() {
+        List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
+        for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
+            for (Subscription sub : entry.getValue().values()) {
+                allSubscriptions.add(sub.asClientTopicCouple());
+            }
+        }
+        return allSubscriptions;
+    }
+
+    @Override
+    public Subscription getSubscription(ClientTopicCouple couple) {
+        Map<Topic, Subscription> subscriptions = m_persistentSubscriptions.get(couple.clientID);
+        if (subscriptions == null || subscriptions.isEmpty()) {
+            return null;
+        }
+        return subscriptions.get(couple.topicFilter);
+    }
+
+    @Override
+    public List<Subscription> getSubscriptions() {
+        List<Subscription> subscriptions = new ArrayList<>();
+        for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
+            subscriptions.addAll(entry.getValue().values());
+        }
+        return subscriptions;
+    }
+
+    @Override
+    public void inFlightAck(String clientID, int messageID) {
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        if (m == null) {
+            LOG.error("Can't find the inFlight record for client <{}>", clientID);
+            return;
+        }
+        m.remove(messageID);
+    }
+
+    @Override
+    public void inFlight(String clientID, int messageID, MessageGUID guid) {
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        if (m == null) {
+            m = new HashMap<>();
+        }
+        m.put(messageID, guid);
+        this.m_inflightStore.put(clientID, m);
+    }
+
+    /**
+     * Return the next valid packetIdentifier for the given client session.
+     */
+    @Override
+    public int nextPacketID(String clientID) {
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        if (m == null) {
+            m = new HashMap<>();
+            int nextPacketId = 1;
+            m.put(nextPacketId, null);
+            return nextPacketId;
+        }
+        int maxId = m.keySet().isEmpty() ? 0 :Collections.max(m.keySet());
+        int nextPacketId = (maxId + 1) % 0xFFFF;
+        m.put(nextPacketId, null);
+        return nextPacketId;
+    }
+
+    @Override
+    public BlockingQueue<StoredMessage> queue(String clientID) {
+        final ArrayBlockingQueue<StoredMessage> emptyQueue = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE);
+        BlockingQueue<StoredMessage> messagesQueue = Utils.defaultGet(queues, clientID, emptyQueue);
+        queues.put(clientID, messagesQueue);
+        return messagesQueue;
+    }
+
+    @Override
+    public void dropQueue(String clientID) {
+        queues.remove(clientID);
+    }
+
+    @Override
+    public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
+        LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID);
+        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
+        if (m == null) {
+            LOG.error("Can't find the inFlight record for client <{}>", clientID);
+            return;
+        }
+        MessageGUID guid = m.remove(messageID);
+
+        LOG.info("Moving to second phase store");
+        final HashMap<Integer, MessageGUID> emptyGuids = new HashMap<>();
+        Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids);
+        messageIDs.put(messageID, guid);
+        m_secondPhaseStore.put(clientID, messageIDs);
+    }
+
+    @Override
+    public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
+        final HashMap<Integer, MessageGUID> emptyGuids = new HashMap<>();
+        Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids);
+        MessageGUID guid = messageIDs.remove(messageID);
+        m_secondPhaseStore.put(clientID, messageIDs);
+        return guid;
+    }
+
+    @Override
+    public StoredMessage getInflightMessage(String clientID, int messageID) {
+        Map<Integer, MessageGUID> inflightMessages = m_inflightStore.get(clientID);
+        MessageGUID guid = inflightMessages.get(messageID);
+        return m_messagesStore.getMessageByGuid(guid);
+    }
+
+    @Override
+    public int getInflightMessagesNo(String clientID) {
+        Map<Integer, MessageGUID> inflightMessages = m_inflightStore.get(clientID);
+        if (inflightMessages == null)
+            return 0;
+        else
+            return inflightMessages.size();
+    }
+
+    @Override
+    public int getPendingPublishMessagesNo(String clientID) {
+        return m_messagesStore.getPendingPublishMessages(clientID);
+    }
+
+    @Override
+    public int getSecondPhaseAckPendingMessages(String clientID) {
+        Map<Integer, MessageGUID> pendingAcks = m_secondPhaseStore.get(clientID);
+        if (pendingAcks == null)
+            return 0;
+        else
+            return pendingAcks.size();
+    }
+
+}

broker/src/main/java/io/moquette/persistence/MemoryStorageService.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2012-2017 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.persistence;
+
+import io.moquette.spi.IMessagesStore;
+import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.IStore;
+
+public class MemoryStorageService implements IStore {
+
+    private MemorySessionStore m_sessionsStore;
+    private MemoryMessagesStore m_messagesStore;
+
+    public MemoryStorageService() {
+        m_messagesStore = new MemoryMessagesStore();
+        m_sessionsStore = new MemorySessionStore(m_messagesStore);
+        m_messagesStore.initStore();
+        m_sessionsStore.initStore();
+    }
+
+    @Override
+    public IMessagesStore messagesStore() {
+        return m_messagesStore;
+    }
+
+    @Override
+    public ISessionsStore sessionsStore() {
+        return m_sessionsStore;
+    }
+
+    @Override
+    public void close() {
+    }
+}

broker/src/main/java/io/moquette/persistence/PersistentSession.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2012-2017 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.persistence;
+
+import java.io.Serializable;
+
+/**
+ * This is a DTO used to persist minimal status (clean session and activation status) of a
+ * session.
+ */
+public class PersistentSession implements Serializable {
+
+    private static final long serialVersionUID = 5052054783220481854L;
+    public final boolean cleanSession;
+
+    public PersistentSession(boolean cleanSession) {
+        this.cleanSession = cleanSession;
+    }
+}

broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java

@@ -54,11 +54,8 @@ public class MQTTMessageLogger extends ChannelDuplexHandler {
                 break;
             case SUBSCRIBE:
                 MqttSubscribeMessage subscribe = (MqttSubscribeMessage) msg;
-                LOG.info(
-                        "{} SUBSCRIBE <{}> to topics {}",
-                        direction,
-                        clientID,
-                        subscribe.payload().topicSubscriptions());
+                LOG.info("{} SUBSCRIBE <{}> to topics {}", direction, clientID,
+                    subscribe.payload().topicSubscriptions());
                 break;
             case UNSUBSCRIBE:
                 MqttUnsubscribeMessage unsubscribe = (MqttUnsubscribeMessage) msg;

broker/src/main/java/io/moquette/spi/ClientSession.java

@@ -74,7 +74,7 @@ public class ClientSession {
      * @return the list of messages to be delivered for client related to the session.
      */
     public BlockingQueue<IMessagesStore.StoredMessage> queue() {
-        LOG.info("Retrieving stored messages. MqttClientId = {}.", clientID);
+        LOG.info("Retrieving enqueued messages. ClientId={}", clientID);
         return this.m_sessionsStore.queue(clientID);
     }
 
@@ -85,14 +85,14 @@ public class ClientSession {
 
     public boolean subscribe(Subscription newSubscription) {
         LOG.info(
-                "Adding new subscription. MqttClientId = {}, topics = {}, qos = {}.",
+                "Adding new subscription. ClientId={}, topics={}, qos={}",
                 newSubscription.getClientId(),
                 newSubscription.getTopicFilter(),
                 newSubscription.getRequestedQos());
         boolean validTopic = newSubscription.getTopicFilter().isValid();
         if (!validTopic) {
             LOG.warn(
-                    "The topic filter is not valid. MqttClientId = {}, topics = {}.",
+                    "The topic filter is not valid. ClientId={}, topics={}",
                     newSubscription.getClientId(),
                     newSubscription.getTopicFilter());
             // send SUBACK with 0x80 for this topic filter
@@ -103,13 +103,9 @@ public class ClientSession {
         // update the selected subscriptions if not present or if has a greater qos
         if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) {
             if (existingSub != null) {
-                LOG.info(
-                        "The subscription already existed with a lower QoS value. It will be updated. "
-                        + "MqttClientId = {}, topics = {}, existingQos = {}, newQos = {}.",
-                        newSubscription.getClientId(),
-                        newSubscription.getTopicFilter(),
-                        existingSub.getRequestedQos(),
-                        newSubscription.getRequestedQos());
+                LOG.info("Subscription already existed with a lower QoS value. It will be updated. ClientId={}, " +
+                    "topics={}, existingQos={}, newQos={}", newSubscription.getClientId(),
+                    newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos());
                 subscriptions.remove(newSubscription);
             }
             subscriptions.add(newSubscription);
@@ -119,7 +115,7 @@ public class ClientSession {
     }
 
     public void unsubscribeFrom(Topic topicFilter) {
-        LOG.info("Removing subscription. MqttClientID = {}, topics = {}.", clientID, topicFilter);
+        LOG.info("Removing subscription. ClientID={}, topics={}", clientID, topicFilter);
         m_sessionsStore.removeSubscription(topicFilter, clientID);
         Set<Subscription> subscriptionsToRemove = new HashSet<>();
         for (Subscription sub : this.subscriptions) {
@@ -132,18 +128,18 @@ public class ClientSession {
 
     public void disconnect() {
         if (this.cleanSession) {
-            LOG.info("The client has disconnected. Removing its subscriptions. MqttClientId = {}.", clientID);
+            LOG.info("Client disconnected. Removing its subscriptions. ClientId={}", clientID);
             // cleanup topic subscriptions
             cleanSession();
         }
     }
 
     public void cleanSession() {
-        LOG.info("Wiping existing subscriptions. MqttClientId = {}.", this.clientID);
+        LOG.info("Wiping existing subscriptions. ClientId={}", this.clientID);
         m_sessionsStore.wipeSubscriptions(this.clientID);
 
         // remove also the messages stored of type QoS1/2
-        LOG.info("Removing stored messages with QoS 1 and 2. MqttClientId = {}.", this.clientID);
+        LOG.info("Removing stored messages with QoS 1 and 2. ClientId={}", this.clientID);
         messagesStore.dropMessagesInSession(this.clientID);
     }
 

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

@@ -27,13 +27,8 @@ class InternalRepublisher {
             // fire as retained the message
             Integer packetID = storedMsg.getQos() == MqttQoS.AT_MOST_ONCE ? null : targetSession.nextPacketId();
             if (packetID != null) {
-                LOG.debug(
-                        "Adding message to inflight zone. "
-                        + "MqttClientId = {}, packetId = {}, messageId = {}, topic = {}.",
-                        targetSession.clientID,
-                        packetID,
-                        storedMsg.getMessageID(),
-                        storedMsg.getTopic());
+                LOG.debug("Adding message to inflight zone. ClientId={}, packetId={}, messageId={}, topic={}",
+                    targetSession.clientID, packetID, storedMsg.getMessageID(), storedMsg.getTopic());
                 targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID);
             }
             MqttPublishMessage publishMsg = retainedPublish(storedMsg);
@@ -51,12 +46,8 @@ class InternalRepublisher {
 
         for (IMessagesStore.StoredMessage pubEvt : storedPublishes) {
             // put in flight zone
-            LOG.debug(
-                    "Adding message ot inflight zone. MqttClientId = {}, guid = {}, messageId = {}, topic = {}.",
-                    clientSession.clientID,
-                    pubEvt.getGuid(),
-                    pubEvt.getMessageID(),
-                    pubEvt.getTopic());
+            LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, messageId={}, topic={}",
+                clientSession.clientID, pubEvt.getGuid(), pubEvt.getMessageID(), pubEvt.getTopic());
             clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
             MqttPublishMessage publishMsg = notRetainedPublish(pubEvt);
             // set the PacketIdentifier only for QoS > 0
@@ -68,39 +59,23 @@ class InternalRepublisher {
     }
 
     private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage, Integer messageID) {
-        return createPublishForQos(
-                storedMessage.getTopic(),
-                storedMessage.getQos(),
-                storedMessage.getMessage(),
-                false,
-                messageID);
+        return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false,
+            messageID);
     }
 
     private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) {
-        return createPublishForQos(
-                storedMessage.getTopic(),
-                storedMessage.getQos(),
-                storedMessage.getMessage(),
-                false,
-                0);
+        return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false,
+            0);
     }
 
     private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) {
-        return createPublishForQos(
-                storedMessage.getTopic(),
-                storedMessage.getQos(),
-                storedMessage.getMessage(),
-                true,
-                0);
+        return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true,
+            0);
     }
 
     private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage, Integer packetID) {
-        return createPublishForQos(
-                storedMessage.getTopic(),
-                storedMessage.getQos(),
-                storedMessage.getMessage(),
-                true,
-                packetID);
+        return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true,
+            packetID);
     }
 
     public static MqttPublishMessage createPublishForQos(String topic, MqttQoS qos, ByteBuf message, boolean retained,

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

@@ -80,12 +80,8 @@ class MessagesPublisher {
                 this.messageSender.sendPublish(targetSession, publishMsg);
             } else {
                 if (!targetSession.isCleanSession()) {
-                    LOG.debug(
-                            "Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}",
-                            pubMsg.getMessageID(),
-                            sub.getClientId(),
-                            sub.getTopicFilter(),
-                            qos);
+                    LOG.debug("Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}",
+                        pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos);
                     // store the message in targetSession queue to deliver
                     targetSession.enqueue(pubMsg);
                 }

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

@@ -24,12 +24,8 @@ class PersistentQueueMessageSender {
         final int messageId = pubMessage.variableHeader().messageId();
         final String topicName = pubMessage.variableHeader().topicName();
         if (LOG.isDebugEnabled()) {
-            LOG.debug(
-                    "Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}",
-                    messageId,
-                    clientId,
-                    topicName,
-                    DebugUtils.payload2Str(pubMessage.payload()));
+            LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId, 
+                clientId, topicName, DebugUtils.payload2Str(pubMessage.payload()));
         } else {
             LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName);
         }
@@ -38,24 +34,12 @@ class PersistentQueueMessageSender {
 
         MqttQoS qos = pubMessage.fixedHeader().qosLevel();
         if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) {
-            LOG.warn(
-                    "The PUBLISH message could not be delivered. It will be stored. "
-                    + "MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.",
-                    messageId,
-                    clientId,
-                    topicName,
-                    qos,
-                    false);
+            LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, qos={}, " +
+                "cleanSession={}", messageId, clientId, topicName, qos, false);
             clientsession.enqueue(asStoredMessage(pubMessage));
         } else {
-            LOG.warn(
-                    "The PUBLISH message could not be delivered. It will be discarded. "
-                    + "MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.",
-                    messageId,
-                    clientId,
-                    topicName,
-                    qos,
-                    true);
+            LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, topic={}, " +
+                "qos={}, cleanSession={}", messageId, clientId, topicName, qos, true);
         }
     }
 }

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

@@ -209,7 +209,7 @@ public class ProtocolProcessor {
         this.allowZeroByteClientId = allowZeroByteClientId;
         m_authorizator = authorizator;
         if (LOG.isDebugEnabled()) {
-            LOG.debug("Initial subscriptions tree={}.", subscriptions.dumpTree());
+            LOG.debug("Initial subscriptions tree={}", subscriptions.dumpTree());
         }
         m_authenticator = authenticator;
         m_messagesStore = storageService;
@@ -302,7 +302,7 @@ public class ProtocolProcessor {
             channel.close();
         }
 
-        LOG.info("The CONNECT message has been processed. CId={}, username={}.", clientId, payload.userName());
+        LOG.info("The CONNECT message has been processed. CId={}, username={}", clientId, payload.userName());
     }
 
     private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) {
@@ -347,7 +347,7 @@ public class ProtocolProcessor {
     }
 
     private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, final String clientId) {
-        LOG.info("Sending connect ACK. CId={}.", clientId);
+        LOG.info("Sending connect ACK. CId={}", clientId);
         final boolean success = descriptor.assignState(DISCONNECTED, SENDACK);
         if (!success) {
             return false;
@@ -363,17 +363,17 @@ public class ProtocolProcessor {
         }
 
         if (isSessionAlreadyStored) {
-            LOG.info("Cleaning session. CId={}.", clientId);
+            LOG.info("Cleaning session. CId={}", clientId);
             clientSession.cleanSession(msg.variableHeader().isCleanSession());
         }
         descriptor.writeAndFlush(okResp);
-        LOG.info("The connect ACK has been sent. CId={}.", clientId);
+        LOG.info("The connect ACK has been sent. CId={}", clientId);
         return true;
     }
 
     private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, final String clientId) {
         int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
-        LOG.info("Configuring connection. CId={}.", clientId);
+        LOG.info("Configuring connection. CId={}", clientId);
         NettyUtils.keepAlive(channel, keepAlive);
         // session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession());
         NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
@@ -399,7 +399,7 @@ public class ProtocolProcessor {
             WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(),
                     willQos);
             m_willStore.put(clientId, will);
-            LOG.info("MQTT last will and testament has been configured. CId={}.", clientId);
+            LOG.info("MQTT last will and testament has been configured. CId={}", clientId);
         }
     }
 
@@ -416,7 +416,7 @@ public class ProtocolProcessor {
             clientSession = m_sessionsStore.createNewSession(clientId, msg.variableHeader().isCleanSession());
         }
         if (msg.variableHeader().isCleanSession()) {
-            LOG.info("Cleaning session. CId={}.", clientId);
+            LOG.info("Cleaning session. CId={}", clientId);
             clientSession.cleanSession();
         }
         return clientSession;
@@ -453,10 +453,10 @@ public class ProtocolProcessor {
      * Republish QoS1 and QoS2 messages stored into the session for the clientID.
      */
     private void republishStoredInSession(ClientSession clientSession) {
-        LOG.info("Republishing stored publish events. CId={}.", clientSession.clientID);
+        LOG.info("Republishing stored publish events. CId={}", clientSession.clientID);
         BlockingQueue<StoredMessage> publishedEvents = clientSession.queue();
         if (publishedEvents.isEmpty()) {
-            LOG.info("There are no stored publish events. ClientId={}.", clientSession.clientID);
+            LOG.info("There are no stored publish events to ClientId={}", clientSession.clientID);
             return;
         }
 
@@ -531,7 +531,7 @@ public class ProtocolProcessor {
     public void internalPublish(MqttPublishMessage msg, final String clientId) {
         final MqttQoS qos = msg.fixedHeader().qosLevel();
         final Topic topic = new Topic(msg.variableHeader().topicName());
-        LOG.info("Sending PUBLISH message. Topic={}, qos={}.", topic, qos);
+        LOG.info("Sending PUBLISH message. Topic={}, qos={}", topic, qos);
 
         MessageGUID guid = null;
         IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
@@ -579,7 +579,7 @@ public class ProtocolProcessor {
         Topic topic = new Topic(tobeStored.getTopic());
         List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
 
-        LOG.info("Publishing will message. CId={}, messageId={}, topic={}.", clientID, messageId, topic);
+        LOG.info("Publishing will message. CId={}, messageId={}, topic={}", clientID, messageId, topic);
         this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions);
     }
 
@@ -610,7 +610,7 @@ public class ProtocolProcessor {
         int messageID = messageId(msg);
         targetSession.moveInFlightToSecondPhaseAckWaiting(messageID);
         // once received a PUBREC reply with a PUBREL(messageID)
-        LOG.debug("Processing PUBREC message. CId={}, messageId={}.", clientID, messageID);
+        LOG.debug("Processing PUBREC message. CId={}, messageId={}", clientID, messageID);
 
         MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0);
         MqttMessage pubRelMessage = new MqttMessage(pubRelHeader, from(messageID));
@@ -620,7 +620,7 @@ public class ProtocolProcessor {
     public void processPubComp(Channel channel, MqttMessage msg) {
         String clientID = NettyUtils.clientID(channel);
         int messageID = messageId(msg);
-        LOG.debug("Processing PUBCOMP message. CId={}, messageId={}.", clientID, messageID);
+        LOG.debug("Processing PUBCOMP message. CId={}, messageId={}", clientID, messageID);
         // once received the PUBCOMP then remove the message from the temp memory
         ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
         StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID);
@@ -631,7 +631,7 @@ public class ProtocolProcessor {
 
     public void processDisconnect(Channel channel) throws InterruptedException {
         final String clientID = NettyUtils.clientID(channel);
-        LOG.info("Processing DISCONNECT message. CId={}.", clientID);
+        LOG.info("Processing DISCONNECT message. CId={}", clientID);
         channel.flush();
         final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID);
         if (existingDescriptor == null) {
@@ -642,42 +642,42 @@ public class ProtocolProcessor {
 
         if (existingDescriptor.doesNotUseChannel(channel)) {
             // another client saved it's descriptor, exit
-            LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}.", clientID);
+            LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}", clientID);
             existingDescriptor.abort();
             return;
         }
 
         if (!removeSubscriptions(existingDescriptor, clientID)) {
-            LOG.warn("Unable to remove subscriptions. Closing connection. CId={}.", clientID);
+            LOG.warn("Unable to remove subscriptions. Closing connection. CId={}", clientID);
             existingDescriptor.abort();
             return;
         }
 
         if (!dropStoredMessages(existingDescriptor, clientID)) {
-            LOG.warn("Unable to drop stored messages. Closing connection. CId={}.", clientID);
+            LOG.warn("Unable to drop stored messages. Closing connection. CId={}", clientID);
             existingDescriptor.abort();
             return;
         }
 
         if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) {
-            LOG.warn("Unable to drop will message. Closing connection. CId={}.", clientID);
+            LOG.warn("Unable to drop will message. Closing connection. CId={}", clientID);
             existingDescriptor.abort();
             return;
         }
 
         if (!existingDescriptor.close()) {
-            LOG.info("The connection has been closed. CId={}.", clientID);
+            LOG.info("The connection has been closed. CId={}", clientID);
             return;
         }
 
         boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor);
         if (!stillPresent) {
             // another descriptor was inserted
-            LOG.warn("Another descriptor has been inserted. CId={}.", clientID);
+            LOG.warn("Another descriptor has been inserted. CId={}", clientID);
             return;
         }
 
-        LOG.info("The DISCONNECT message has been processed. CId={}.", clientID);
+        LOG.info("The DISCONNECT message has been processed. CId={}", clientID);
     }
 
     private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) {
@@ -687,9 +687,9 @@ public class ProtocolProcessor {
         }
 
         if (descriptor.cleanSession) {
-            LOG.info("Removing saved subscriptions. CId={}.", descriptor.clientID);
+            LOG.info("Removing saved subscriptions. CId={}", descriptor.clientID);
             m_sessionsStore.wipeSubscriptions(clientID);
-            LOG.info("The saved subscriptions have been removed. CId={}.", descriptor.clientID);
+            LOG.info("The saved subscriptions have been removed. CId={}", descriptor.clientID);
         }
         return true;
     }
@@ -701,9 +701,9 @@ public class ProtocolProcessor {
         }
 
         if (descriptor.cleanSession) {
-            LOG.debug("Removing messages of session. CId={}.", descriptor.clientID);
+            LOG.debug("Removing messages of session. CId={}", descriptor.clientID);
             this.m_sessionsStore.dropQueue(clientID);
-            LOG.debug("The messages of the session have been removed. CId={}.", descriptor.clientID);
+            LOG.debug("The messages of the session have been removed. CId={}", descriptor.clientID);
         }
         return true;
     }
@@ -714,7 +714,7 @@ public class ProtocolProcessor {
             return false;
         }
 
-        LOG.info("Removing will message. ClientId={}.", descriptor.clientID);
+        LOG.info("Removing will message. ClientId={}", descriptor.clientID);
         // cleanup the will store
         m_willStore.remove(clientID);
         String username = descriptor.getUsername();
@@ -723,7 +723,7 @@ public class ProtocolProcessor {
     }
 
     public void processConnectionLost(String clientID, Channel channel) {
-        LOG.info("Processing connection lost event. CId={}.", clientID);
+        LOG.info("Processing connection lost event. CId={}", clientID);
         ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true);
         connectionDescriptors.removeConnection(oldConnDescr);
         // publish the Will message (if any) for the clientID
@@ -750,7 +750,7 @@ public class ProtocolProcessor {
         List<String> topics = msg.payload().topics();
         String clientID = NettyUtils.clientID(channel);
 
-        LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}.", clientID, topics);
+        LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}", clientID, topics);
 
         ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
         for (String t : topics) {
@@ -764,7 +764,7 @@ public class ProtocolProcessor {
                 return;
             }
 
-            LOG.debug("Removing subscription. CId={}, topic={}.", clientID, topic);
+            LOG.debug("Removing subscription. CId={}, topic={}", clientID, topic);
             subscriptions.removeSubscription(topic, clientID);
             clientSession.unsubscribeFrom(topic);
             String username = NettyUtils.userName(channel);
@@ -776,29 +776,28 @@ public class ProtocolProcessor {
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0);
         MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID));
 
-        LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}.", clientID, topics, messageID);
+        LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}", clientID, topics, messageID);
         channel.writeAndFlush(ackMessage);
     }
 
     public void processSubscribe(Channel channel, MqttSubscribeMessage msg) {
         String clientID = NettyUtils.clientID(channel);
         int messageID = messageId(msg);
-        LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}.", clientID, messageID);
+        LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}", clientID, messageID);
 
         RunningSubscription executionKey = new RunningSubscription(clientID, messageID);
-        SubscriptionState currentStatus = this.subscriptionInCourse
-                .putIfAbsent(executionKey, SubscriptionState.VERIFIED);
+        SubscriptionState currentStatus = subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED);
         if (currentStatus != null) {
-            LOG.warn("Client sent another SUBSCRIBE message while this one was being processed " +
-                "CId={}, messageId={}", clientID, messageID);
+            LOG.warn("Client sent another SUBSCRIBE message while this one was being processed CId={}, messageId={}",
+                clientID, messageID);
             return;
         }
         String username = NettyUtils.userName(channel);
         List<MqttTopicSubscription> ackTopics = doVerify(clientID, username, msg);
         MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID);
         if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) {
-            LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified " +
-                "CId={}, messageId={}", clientID, messageID);
+            LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified CId={}, " +
+                "messageId={}", clientID, messageID);
             return;
         }
 
@@ -861,19 +860,18 @@ public class ProtocolProcessor {
             Topic topic = new Topic(req.topicName());
             if (!m_authorizator.canRead(topic, username, clientSession.clientID)) {
                 // send SUBACK with 0x80, the user hasn't credentials to read the topic
-                LOG.error("Client does not have read permissions on the topic " +
-                    "CId={}, username={}, messageId={}, topic={}",
-                        clientID, username, messageId, topic);
+                LOG.error("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.info("Client will be subscribed to the topic " +
-                        "CId={}, username={}, messageId={}, topic={}", clientID, username, messageId, topic);
+                    LOG.info("Client will be subscribed to the topic CId={}, username={}, messageId={}, topic={}",
+                        clientID, username, messageId, topic);
                     qos = req.qualityOfService();
                 } else {
-                    LOG.error("Topic filter is not valid CId={}, username={}, messageId={}, topic={}",
-                            clientID, username, messageId, topic);
+                    LOG.error("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", clientID,
+                        username, messageId, topic);
                     qos = FAILURE;
                 }
                 ackTopics.add(new MqttTopicSubscription(topic.toString(), qos));

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

@@ -32,7 +32,6 @@ import io.moquette.spi.impl.security.PermitAllAuthorizator;
 import io.moquette.spi.impl.security.ResourceAuthenticator;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
-import io.moquette.spi.persistence.MapDBPersistentStore;
 import io.moquette.spi.security.IAuthenticator;
 import io.moquette.spi.security.IAuthorizator;
 import org.slf4j.Logger;
@@ -51,6 +50,7 @@ import java.util.List;
 public class ProtocolProcessorBootstrapper {
 
     private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessorBootstrapper.class);
+    public static final String MAPDB_STORE_CLASS = "io.moquette.persistence.MemoryStorageService";
 
     private SubscriptionsStore subscriptions;
 
@@ -83,7 +83,7 @@ public class ProtocolProcessorBootstrapper {
      *            an implementation of the authorizator to be used, if null load that specified in
      *            config and fallback on the default one (permit all).
      * @param server
-     *            the serber to init.
+     *            the server to init.
      * @return the processor created for the broker.
      */
     public ProtocolProcessor init(IConfig props, List<? extends InterceptHandler> embeddedObservers,
@@ -92,31 +92,21 @@ public class ProtocolProcessorBootstrapper {
 
         IMessagesStore messagesStore;
         LOG.info("Initializing messages and sessions stores...");
-        String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, "");
-        if (storageClassName != null && !storageClassName.isEmpty()) {
-            final IStore store = loadClass(storageClassName, IStore.class, Server.class, server);
-            messagesStore = store.messagesStore();
-            m_sessionsStore = store.sessionsStore();
-            storeShutdown = new Runnable() {
-
-                @Override
-                public void run() {
-                    store.close();
-                }
-            };
-        } else {
-            final MapDBPersistentStore m_mapStorage = new MapDBPersistentStore(props, server.getScheduler());
-            m_mapStorage.initStore();
-            messagesStore = m_mapStorage.messagesStore();
-            m_sessionsStore = m_mapStorage.sessionsStore();
-            storeShutdown = new Runnable() {
-
-                @Override
-                public void run() {
-                    m_mapStorage.close();
-                }
-            };
+        String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, MAPDB_STORE_CLASS);
+        if (storageClassName == null || storageClassName.isEmpty()) {
+            LOG.error("storage_class property not defined");
+            throw new IllegalArgumentException("Can't start a valid persistence layer");
         }
+        final IStore store = loadClass(storageClassName, IStore.class, Server.class, server);
+        messagesStore = store.messagesStore();
+        m_sessionsStore = store.sessionsStore();
+        storeShutdown = new Runnable() {
+
+            @Override
+            public void run() {
+                store.close();
+            }
+        };
 
         LOG.info("Configuring message interceptors...");
 
@@ -148,7 +138,7 @@ public class ProtocolProcessorBootstrapper {
             } else {
                 authenticator = new ResourceAuthenticator(resourceLoader, passwdPath);
             }
-            LOG.info("An {} authenticator instance will be used.", authenticator.getClass().getName());
+            LOG.info("An {} authenticator instance will be used", authenticator.getClass().getName());
         }
 
         LOG.info("Configuring MQTT authorizator...");
@@ -162,119 +152,78 @@ public class ProtocolProcessorBootstrapper {
             if (aclFilePath != null && !aclFilePath.isEmpty()) {
                 authorizator = new DenyAllAuthorizator();
                 try {
-                    LOG.info("Parsing ACL file. Path = {}.", aclFilePath);
+                    LOG.info("Parsing ACL file. Path = {}", aclFilePath);
                     authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath));
                 } catch (ParseException pex) {
-                    LOG.error(
-                            "Unable to parse ACL file. Path = {}, cause = {}, errorMessage = {}.",
-                            aclFilePath,
-                            pex.getCause(),
-                            pex.getMessage());
+                    LOG.error("Unable to parse ACL file. path="+ aclFilePath, pex);
                 }
             } else {
                 authorizator = new PermitAllAuthorizator();
             }
-            LOG.info("An {} authorizator instance will be used.", authorizator.getClass().getName());
+            LOG.info("An {} authorizator instance will be used", authorizator.getClass().getName());
         }
 
         LOG.info("Initializing connection descriptor store...");
         connectionDescriptors = new ConnectionDescriptorStore(m_sessionsStore);
 
         LOG.info("Initializing MQTT protocol processor...");
-        boolean allowAnonymous = Boolean
-                .parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
-        boolean allowZeroByteClientId = Boolean
-                .parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
-        m_processor.init(
-                connectionDescriptors,
-                subscriptions,
-                messagesStore,
-                m_sessionsStore,
-                authenticator,
-                allowAnonymous,
-                allowZeroByteClientId,
-                authorizator,
-                m_interceptor,
+        boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
+        boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
+        m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator,
+                allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor,
                 props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
         return m_processor;
     }
 
     @SuppressWarnings("unchecked")
-    private <T, U> T loadClass(String className, Class<T> iface, Class<U> constructorArgClass, U props) {
+    private <T, U> T loadClass(String className, Class<T> intrface, Class<U> constructorArgClass, U props) {
         T instance = null;
         try {
-            LOG.info("Loading class. ClassName = {}, interfaceName = {}.", className, iface.getName());
+            LOG.info("Loading class. ClassName={}, interfaceName={}", className, intrface.getName());
             Class<?> clazz = Class.forName(className);
 
             // check if method getInstance exists
             Method method = clazz.getMethod("getInstance", new Class[]{});
             try {
-                LOG.info(
-                        "Invoking getInstance() method. ClassName = {}, interfaceName = {}.",
-                        className,
-                        iface.getName());
+                LOG.info("Invoking getInstance() method. ClassName = {}, interfaceName = {}.", className, intrface.getName());
                 instance = (T) method.invoke(null, new Object[]{});
             } catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) {
-                LOG.error(
-                        "Unable to invoke getInstance() method. "
-                        + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
-                        className,
-                        iface.getName(),
-                        ex.getCause(),
-                        ex.getMessage());
+                LOG.error("Unable to invoke getInstance() method. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
+                        className, intrface.getName(), ex.getCause(), ex.getMessage());
                 return null;
             }
         } catch (NoSuchMethodException nsmex) {
             try {
                 // check if constructor with constructor arg class parameter
                 // exists
-                LOG.info(
-                        "Invoking constructor with {} argument. ClassName = {}, interfaceName = {}.",
-                        constructorArgClass.getName(),
-                        className,
-                        iface.getName());
-                instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface)
-                        .getConstructor(constructorArgClass).newInstance(props);
+                LOG.info("Invoking constructor with {} argument. ClassName = {}, interfaceName = {}.",
+                        constructorArgClass.getName(), className, intrface.getName());
+                instance = this.getClass().getClassLoader()
+                    .loadClass(className)
+                    .asSubclass(intrface)
+                    .getConstructor(constructorArgClass)
+                    .newInstance(props);
             } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
-                LOG.warn(
-                        "Unable to invoke constructor with {} argument. "
-                        + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
-                        constructorArgClass.getName(),
-                        className,
-                        iface.getName(),
-                        ex.getCause(),
-                        ex.getMessage());
+                LOG.warn("Unable to invoke constructor with {} argument. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
+                        constructorArgClass.getName(), className, intrface.getName(), ex.getCause(), ex.getMessage());
                 return null;
             } catch (NoSuchMethodException | InvocationTargetException e) {
                 try {
-                    LOG.info(
-                            "Invoking default constructor. ClassName = {}, interfaceName = {}.",
-                            className,
-                            iface.getName());
+                    LOG.info("Invoking default constructor. ClassName = {}, interfaceName = {}.", className, intrface.getName());
                     // fallback to default constructor
-                    instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface).newInstance();
+                    instance = this.getClass().getClassLoader().loadClass(className).asSubclass(intrface).newInstance();
                 } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
-                    LOG.error(
-                            "Unable to invoke default constructor. "
-                            + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
-                            className,
-                            iface.getName(),
-                            ex.getCause(),
-                            ex.getMessage());
+                    LOG.error("Unable to invoke default constructor. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
+                            className, intrface.getName(), ex.getCause(), ex.getMessage());
                     return null;
                 }
             }
         } catch (ClassNotFoundException ex) {
-            LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, iface.getName());
+            LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, intrface.getName());
             return null;
         } catch (SecurityException ex) {
-            LOG.error(
-                    "Unable to load class due to a security violation. "
-                    + "ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
-                    className,
-                    iface.getName(),
-                    ex.getCause(),
-                    ex.getMessage());
+            LOG.error("Unable to load class due to a security violation. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.",
+                    className, intrface.getName(), ex.getCause(), ex.getMessage());
             return null;
         }
 

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

@@ -47,20 +47,12 @@ class Qos0PublishHandler extends QosPublishHandler {
         final int messageID = msg.variableHeader().messageId();
 
         if (LOG.isTraceEnabled()) {
-            LOG.trace(
-                    "Sending publish message to subscribers. "
-                    + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
-                    clientID,
-                    topic,
-                    messageID,
-                    DebugUtils.payload2Str(toStoreMsg.getMessage()),
-                    subscriptions.dumpTree());
+            LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
+                "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
+                subscriptions.dumpTree());
         } else {
-            LOG.info(
-                    "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.",
-                    clientID,
-                    topic,
-                    messageID);
+            LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic,
+                messageID);
         }
 
         List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);

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

@@ -57,37 +57,26 @@ class Qos1PublishHandler extends QosPublishHandler {
         toStoreMsg.setClientID(clientID);
 
         if (LOG.isTraceEnabled()) {
-            LOG.trace(
-                    "Sending publish message to subscribers. "
-                    + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
-                    clientID,
-                    topic,
-                    messageID,
-                    DebugUtils.payload2Str(toStoreMsg.getMessage()),
-                    subscriptions.dumpTree());
+            LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
+                "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
+                subscriptions.dumpTree());
         } else {
-            LOG.info(
-                    "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.",
-                    clientID,
-                    topic,
-                    messageID);
+            LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic,
+                messageID);
         }
 
         List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
         this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
 
-        // send PUBACK
-        // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) {
         sendPubAck(clientID, messageID);
-        // }
 
         if (msg.fixedHeader().isRetain()) {
             if (!msg.payload().isReadable()) {
                 m_messagesStore.cleanRetained(topic);
             } else {
                 // before wasn't stored
-                MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
-                m_messagesStore.storeRetained(topic, guid);
+                //MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
+                m_messagesStore.storeRetained(topic, toStoreMsg.getGuid());
             }
         }
 
@@ -102,17 +91,13 @@ class Qos1PublishHandler extends QosPublishHandler {
 
         try {
             if (connectionDescriptors == null) {
-                throw new RuntimeException(
-                        "Internal bad error, found connectionDescriptors to null while it should be initialized, "
-                        + "somewhere it's overwritten!!");
+                throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should " +
+                    "be initialized, somewhere it's overwritten!!");
             }
             LOG.debug("clientIDs are {}", connectionDescriptors);
             if (!connectionDescriptors.isConnected(clientId)) {
-                throw new RuntimeException(
-                        String.format(
-                                "Can't find a ConnectionDescriptor for client %s in cache %s",
-                                clientId,
-                                connectionDescriptors));
+                throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s",
+                    clientId,connectionDescriptors));
             }
             connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId);
         } catch (Throwable t) {

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

@@ -58,27 +58,19 @@ class Qos2PublishHandler extends QosPublishHandler {
         toStoreMsg.setClientID(clientID);
 
         if (LOG.isTraceEnabled()) {
-            LOG.trace(
-                    "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, "
-                            + "payload = {}, subscriptionTree = {}.",
-                    clientID,
-                    topic,
-                    messageID,
-                    DebugUtils.payload2Str(toStoreMsg.getMessage()),
-                    subscriptions.dumpTree());
+            LOG.trace( "Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
+                "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()),
+                subscriptions.dumpTree());
         } else {
-            LOG.info(
-                    "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.",
-                    clientID,
-                    topic,
-                    messageID);
+            LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic,
+                messageID);
         }
 
         // QoS2
         MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
         // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) {
         sendPubRec(clientID, messageID);
-        // }
+
         // Next the client will send us a pub rel
         // NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
 
@@ -100,27 +92,19 @@ class Qos2PublishHandler extends QosPublishHandler {
     void processPubRel(Channel channel, MqttMessage msg) {
         String clientID = NettyUtils.clientID(channel);
         int messageID = messageId(msg);
-        LOG.info("Processing PUBREL message. MqttClientId = {}, messageId = {}.", clientID, messageID);
+        LOG.info("Processing PUBREL message. ClientId={}, messageId={}", clientID, messageID);
         ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
         IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
         final Topic topic = new Topic(evt.getTopic());
         List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
 
         if (LOG.isDebugEnabled()) {
-            LOG.debug(
-                    "Sending publish message to subscribers. "
-                    + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
-                    clientID,
-                    topic,
-                    messageID,
-                    DebugUtils.payload2Str(evt.getMessage()),
-                    subscriptions.dumpTree());
+            LOG.debug("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " +
+                "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(evt.getMessage()),
+                subscriptions.dumpTree());
         } else {
-            LOG.info(
-                    "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.",
-                    clientID,
-                    topic,
-                    messageID);
+            LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic,
+                messageID);
         }
         this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions);
 
@@ -136,14 +120,14 @@ class Qos2PublishHandler extends QosPublishHandler {
     }
 
     private void sendPubRec(String clientID, int messageID) {
-        LOG.debug("Sending PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID);
+        LOG.debug("Sending PUBREC message. ClientId={}, messageId={}", clientID, messageID);
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE, false, 0);
         MqttMessage pubRecMessage = new MqttMessage(fixedHeader, from(messageID));
         connectionDescriptors.sendMessage(pubRecMessage, messageID, clientID);
     }
 
     private void sendPubComp(String clientID, int messageID) {
-        LOG.debug("Sending PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID);
+        LOG.debug("Sending PUBCOMP message. ClientId={}, messageId={}", clientID, messageID);
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, AT_MOST_ONCE, false, 0);
         MqttMessage pubCompMessage = new MqttMessage(fixedHeader, from(messageID));
         connectionDescriptors.sendMessage(pubCompMessage, messageID, clientID);

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

@@ -47,7 +47,7 @@ public class Topic implements Serializable {
                 valid = true;
             } catch (ParseException e) {
                 valid = false;
-                LOG.error(e.toString(), e);
+                LOG.error("Error parsing the topic: {}, message: {}", topic, e.getMessage());
             }
         }
 

broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java

@@ -1,154 +0,0 @@
-/*
- * Copyright (c) 2012-2017 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.spi.persistence;
-
-import io.moquette.spi.IMatchingCondition;
-import io.moquette.spi.IMessagesStore;
-import io.moquette.spi.MessageGUID;
-import io.moquette.spi.impl.subscriptions.Topic;
-import org.mapdb.DB;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.util.*;
-import java.util.concurrent.ConcurrentMap;
-
-/**
- * IMessagesStore implementation backed by MapDB.
- *
- * @author andrea
- */
-class MapDBMessagesStore implements IMessagesStore {
-
-    private static final Logger LOG = LoggerFactory.getLogger(MapDBMessagesStore.class);
-
-    private DB m_db;
-
-    // maps clientID -> guid
-    private ConcurrentMap<Topic, MessageGUID> m_retainedStore;
-    // maps guid to message, it's message store
-    private ConcurrentMap<MessageGUID, IMessagesStore.StoredMessage> m_persistentMessageStore;
-
-    MapDBMessagesStore(DB db) {
-        m_db = db;
-    }
-
-    @Override
-    public void initStore() {
-        LOG.info("Initializing store...");
-        m_retainedStore = m_db.getHashMap("retained");
-        m_persistentMessageStore = m_db.getHashMap("persistedMessages");
-    }
-
-    @Override
-    public void storeRetained(Topic topic, MessageGUID guid) {
-        LOG.debug("Storing retained messages. Topic = {}, guid = {}.", topic, guid);
-        m_retainedStore.put(topic, guid);
-    }
-
-    @Override
-    public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
-        LOG.debug("Scanning retained messages...");
-        List<StoredMessage> results = new ArrayList<>();
-        for (Map.Entry<Topic, MessageGUID> entry : m_retainedStore.entrySet()) {
-            final MessageGUID guid = entry.getValue();
-            StoredMessage storedMsg = m_persistentMessageStore.get(guid);
-            if (condition.match(entry.getKey())) {
-                results.add(storedMsg);
-            }
-        }
-
-        if (LOG.isTraceEnabled()) {
-            LOG.trace("The retained messages have been scanned. MatchingMessages = {}.", results);
-        }
-
-        return results;
-    }
-
-    @Override
-    public MessageGUID storePublishForFuture(StoredMessage storedMessage) {
-        assert storedMessage.getClientID() != null : "The message to be persisted must have a valid client ID";
-        MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
-        storedMessage.setGuid(guid);
-        LOG.debug(
-                "Storing publish event. MqttClientId = {}, messageId = {}, guid = {}, topic = {}.",
-                storedMessage.getClientID(),
-                storedMessage.getMessageID(),
-                guid,
-                storedMessage.getTopic());
-        m_persistentMessageStore.put(guid, storedMessage);
-        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db
-                .getHashMap(messageId2GuidsMapName(storedMessage.getClientID()));
-        messageIdToGuid.put(storedMessage.getMessageID(), guid);
-        return guid;
-    }
-
-    @Override
-    public void dropMessagesInSession(String clientID) {
-        LOG.debug("Dropping stored messages. ClientId = {}.", clientID);
-        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
-        for (MessageGUID guid : messageIdToGuid.values()) {
-            removeStoredMessage(guid);
-        }
-        messageIdToGuid.clear();
-    }
-
-    void removeStoredMessage(MessageGUID guid) {
-        // remove only the not retained and no more referenced
-        StoredMessage storedMessage = m_persistentMessageStore.get(guid);
-        if (!storedMessage.isRetained()) {
-            LOG.debug(
-                    "Dropping stored message. ClientId = {}, messageId = {}, guid = {}, topic = {}.",
-                    storedMessage.getClientID(),
-                    storedMessage.getMessageID(),
-                    guid,
-                    storedMessage.getTopic());
-            m_persistentMessageStore.remove(guid);
-        }
-    }
-
-    @Override
-    public StoredMessage getMessageByGuid(MessageGUID guid) {
-        LOG.debug("Retrieving stored message. Guid = {}.", guid);
-        return m_persistentMessageStore.get(guid);
-    }
-
-    @Override
-    public void cleanRetained(Topic topic) {
-        LOG.debug("Cleaning retained messages. Topic = {}.", topic);
-        m_retainedStore.remove(topic);
-    }
-
-    @Override
-    public int getPendingPublishMessages(String clientID) {
-        ConcurrentMap<Integer, MessageGUID> messageIdToGuidMap = m_db
-                .getHashMap(messageId2GuidsMapName(clientID));
-        return messageIdToGuidMap.size();
-    }
-
-    @Override
-    public MessageGUID mapToGuid(String clientID, int messageID) {
-        LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID);
-        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
-        MessageGUID result = messageIdToGuid.get(messageID);
-        LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result);
-        return result;
-    }
-
-    static String messageId2GuidsMapName(String clientID) {
-        return "guidsMapping_" + clientID;
-    }
-}

broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java

@@ -1,149 +0,0 @@
-/*
- * Copyright (c) 2012-2017 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.spi.persistence;
-
-import io.moquette.server.config.IConfig;
-import io.moquette.spi.IMessagesStore;
-import io.moquette.spi.ISessionsStore;
-import io.moquette.spi.IStore;
-import org.mapdb.DB;
-import org.mapdb.DBMaker;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import static io.moquette.BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME;
-import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME;
-
-/**
- * MapDB main persistence implementation
- */
-public class MapDBPersistentStore implements IStore {
-
-    /**
-     * This is a DTO used to persist minimal status (clean session and activation status) of a
-     * session.
-     */
-    public static class PersistentSession implements Serializable {
-
-        private static final long serialVersionUID = 5052054783220481854L;
-        public final boolean cleanSession;
-
-        public PersistentSession(boolean cleanSession) {
-            this.cleanSession = cleanSession;
-        }
-    }
-
-    private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
-
-    private DB m_db;
-    private final String m_storePath;
-    private final int m_autosaveInterval; // in seconds
-
-    protected final ScheduledExecutorService m_scheduler;
-    private IMessagesStore m_messageStore;
-    private ISessionsStore m_sessionsStore;
-
-    public MapDBPersistentStore(IConfig props, ScheduledExecutorService scheduler) {
-        this.m_storePath = props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, "");
-        this.m_autosaveInterval = Integer.parseInt(props.getProperty(AUTOSAVE_INTERVAL_PROPERTY_NAME, "30"));
-        this.m_scheduler = scheduler;
-    }
-
-    /**
-     * Factory method to create message store backed by MapDB
-     *
-     * @return the message store instance.
-     */
-    @Override
-    public IMessagesStore messagesStore() {
-        return m_messageStore;
-    }
-
-    @Override
-    public ISessionsStore sessionsStore() {
-        return m_sessionsStore;
-    }
-
-    public void initStore() {
-        LOG.info("Initializing MapDB store...");
-        if (m_storePath == null || m_storePath.isEmpty()) {
-            LOG.warn("The MapDB store file path is empty. Using in-memory store.");
-            m_db = DBMaker.newMemoryDB().make();
-        } else {
-            File tmpFile;
-            try {
-                LOG.info("Using user-defined MapDB store file. Path = {}.", m_storePath);
-                tmpFile = new File(m_storePath);
-                boolean fileNewlyCreated = tmpFile.createNewFile();
-                LOG.warn("Using {} MapDB store file. Path = {}.", fileNewlyCreated ? "fresh" : "existing", m_storePath);
-            } catch (IOException ex) {
-                LOG.error(
-                        "Unable to open MapDB store file. Path = {}, cause = {}, errorMessage = {}.",
-                        m_storePath,
-                        ex.getCause(),
-                        ex.getMessage());
-                throw new RuntimeException(
-                        "Can't create temp file for subscriptions storage [" + m_storePath + "]",
-                        ex);
-            }
-            m_db = DBMaker.newFileDB(tmpFile).make();
-        }
-        LOG.info("Scheduling MapDB commit task...");
-        m_scheduler.scheduleWithFixedDelay(new Runnable() {
-
-            @Override
-            public void run() {
-                LOG.debug("Committing to MapDB...");
-                m_db.commit();
-            }
-        }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS);
-
-        // TODO check m_db is valid and
-        m_messageStore = new MapDBMessagesStore(m_db);
-        m_messageStore.initStore();
-
-        m_sessionsStore = new MapDBSessionsStore(m_db, m_messageStore);
-        m_sessionsStore.initStore();
-    }
-
-    @Override
-    public void close() {
-        if (this.m_db.isClosed()) {
-            LOG.warn("The MapDB store is already closed. Nothing will be done.");
-            return;
-        }
-        LOG.info("Performing last commit to MapDB...");
-        this.m_db.commit();
-        LOG.info("Closing MapDB store...");
-        this.m_db.close();
-        LOG.info("Stopping MapDB commit tasks...");
-        this.m_scheduler.shutdown();
-        try {
-            m_scheduler.awaitTermination(10L, TimeUnit.SECONDS);
-        } catch (InterruptedException e) {
-        }
-        if (!m_scheduler.isTerminated()) {
-            LOG.warn("Forcing shutdown of MapDB commit tasks...");
-            m_scheduler.shutdown();
-        }
-        LOG.info("The MapDB store has been closed successfully.");
-    }
-}

broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java

@@ -1,336 +0,0 @@
-/*
- * Copyright (c) 2012-2017 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.spi.persistence;
-
-import io.moquette.spi.ClientSession;
-import io.moquette.spi.IMessagesStore;
-import io.moquette.spi.IMessagesStore.StoredMessage;
-import io.moquette.spi.ISessionsStore;
-import io.moquette.spi.MessageGUID;
-import io.moquette.spi.impl.Utils;
-import io.moquette.spi.impl.subscriptions.Subscription;
-import io.moquette.spi.impl.subscriptions.Topic;
-import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
-import org.mapdb.DB;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import java.util.*;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
-/**
- * ISessionsStore implementation backed by MapDB.
- *
- * @author andrea
- */
-class MapDBSessionsStore implements ISessionsStore {
-
-    private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class);
-
-    // maps clientID->[MessageId -> guid]
-    private ConcurrentMap<String, ConcurrentMap<Integer, MessageGUID>> m_inflightStore;
-    // map clientID <-> set of currently in flight packet identifiers
-    private Map<String, Set<Integer>> m_inFlightIds;
-    private ConcurrentMap<String, PersistentSession> m_persistentSessions;
-    // maps clientID->[MessageId -> guid]
-    private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
-
-    private final DB m_db;
-    private final IMessagesStore m_messagesStore;
-
-    MapDBSessionsStore(DB db, IMessagesStore messagesStore) {
-        m_db = db;
-        m_messagesStore = messagesStore;
-    }
-
-    @Override
-    public void initStore() {
-        m_inflightStore = m_db.getHashMap("inflight");
-        m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
-        m_persistentSessions = m_db.getHashMap("sessions");
-        m_secondPhaseStore = m_db.getHashMap("secondPhase");
-    }
-
-    @Override
-    public void addNewSubscription(Subscription newSubscription) {
-        LOG.info(
-                "Adding new subscription. CId= {}, topics = {}.",
-                newSubscription.getClientId(),
-                newSubscription.getTopicFilter());
-        final String clientID = newSubscription.getClientId();
-        m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription);
-
-        if (LOG.isTraceEnabled()) {
-            LOG.trace(
-                    "The subscription has been added. CId= {}, topics = {}, clientSubscriptions = {}.",
-                    newSubscription.getClientId(),
-                    newSubscription.getTopicFilter(),
-                    m_db.getHashMap("subscriptions_" + clientID));
-        }
-    }
-
-    @Override
-    public void removeSubscription(Topic topicFilter, String clientID) {
-        LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter);
-        if (!m_db.exists("subscriptions_" + clientID)) {
-            return;
-        }
-        m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter);
-        if (LOG.isDebugEnabled()) {
-            LOG.debug(
-                    "The subscription has been removed. CId= {}, topics = {}, clientSubscriptions = {}.",
-                    clientID,
-                    topicFilter,
-                    m_db.getHashMap("subscriptions_" + clientID));
-        }
-    }
-
-    @Override
-    public void wipeSubscriptions(String clientID) {
-        LOG.info("Wiping subscriptions. CId= {}.", clientID);
-        m_db.delete("subscriptions_" + clientID);
-        if (LOG.isDebugEnabled()) {
-            LOG.debug(
-                    "The subscriptions have been removed. CId= {}, clientSubscriptions = {}.",
-                    clientID,
-                    m_db.getHashMap("subscriptions_" + clientID));
-        }
-    }
-
-    @Override
-    public List<ClientTopicCouple> listAllSubscriptions() {
-        LOG.info("Retrieving existing subscriptions...");
-        final List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
-        for (String clientID : m_persistentSessions.keySet()) {
-            ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
-            for (Topic topicFilter : clientSubscriptions.keySet()) {
-                allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter));
-            }
-        }
-        if (LOG.isTraceEnabled()) {
-            LOG.trace("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions);
-        }
-        return allSubscriptions;
-    }
-
-    @Override
-    public Subscription getSubscription(ClientTopicCouple couple) {
-        ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
-        LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions);
-        return clientSubscriptions.get(couple.topicFilter);
-    }
-
-    @Override
-    public List<Subscription> getSubscriptions() {
-        LOG.info("Retrieving existing subscriptions...");
-        List<Subscription> subscriptions = new ArrayList<>();
-        for (String clientID : m_persistentSessions.keySet()) {
-            ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
-            subscriptions.addAll(clientSubscriptions.values());
-        }
-        LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions);
-        return subscriptions;
-    }
-
-    @Override
-    public boolean contains(String clientID) {
-        return m_db.exists("subscriptions_" + clientID);
-    }
-
-    @Override
-    public ClientSession createNewSession(String clientID, boolean cleanSession) {
-        if (m_persistentSessions.containsKey(clientID)) {
-            LOG.error(
-                    "Unable to create a new session: the client ID is already in use. CId= {}, cleanSession = {}.",
-                    clientID,
-                    cleanSession);
-            throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID);
-        }
-        LOG.info("Creating new session. CId= {}, cleanSession = {}.", clientID, cleanSession);
-        m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession));
-        return new ClientSession(clientID, m_messagesStore, this, cleanSession);
-    }
-
-    @Override
-    public ClientSession sessionForClient(String clientID) {
-        LOG.info("Retrieving session. CId= {}.", clientID);
-        if (!m_persistentSessions.containsKey(clientID)) {
-            LOG.warn("The session does not exist. CId= {}.", clientID);
-            return null;
-        }
-
-        PersistentSession storedSession = m_persistentSessions.get(clientID);
-        return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
-    }
-
-    @Override
-    public Collection<ClientSession> getAllSessions() {
-        Collection<ClientSession> result = new ArrayList<>();
-        for (Map.Entry<String, PersistentSession> entry : m_persistentSessions.entrySet()) {
-            result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession));
-        }
-        return result;
-    }
-
-    @Override
-    public void updateCleanStatus(String clientID, boolean cleanSession) {
-        LOG.info("Updating cleanSession flag. CId= {}, cleanSession = {}.", clientID, cleanSession);
-        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
-    }
-
-    /**
-     * Return the next valid packetIdentifier for the given client session.
-     */
-    @Override
-    public int nextPacketID(String clientID) {
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("Generating next packet ID. CId= {}.", clientID);
-        }
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient == null) {
-            int nextPacketId = 1;
-            inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
-            inFlightForClient.add(nextPacketId);
-            this.m_inFlightIds.put(clientID, inFlightForClient);
-            return nextPacketId;
-
-        }
-
-        int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
-        int nextPacketId = (maxId % 0xFFFF) + 1;
-        inFlightForClient.add(nextPacketId);
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("The next packet ID has been generated. CId= {}, result = {}.", clientID, nextPacketId);
-        }
-        return nextPacketId;
-    }
-
-    @Override
-    public void inFlightAck(String clientID, int messageID) {
-        LOG.debug("Acknowledging inflight message. CId= {}, messageId = {}.", clientID, messageID);
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
-            LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
-            return;
-        }
-        m.remove(messageID);
-
-        // remove from the ids store
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient != null) {
-            inFlightForClient.remove(messageID);
-        }
-    }
-
-    @Override
-    public void inFlight(String clientID, int messageID, MessageGUID guid) {
-        LOG.debug("Storing inflight message. CId= {}, messageId = {}, guid = {}.", clientID, messageID, guid);
-        ConcurrentMap<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
-            m = new ConcurrentHashMap<>();
-        }
-        m.put(messageID, guid);
-        LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid);
-        this.m_inflightStore.put(clientID, m);
-    }
-
-    @Override
-    public BlockingQueue<StoredMessage> queue(String clientID) {
-        LOG.info("Queuing pending message. CId= {}, guid = {}.", clientID);
-        return this.m_db.getQueue(clientID);
-    }
-
-    @Override
-    public void dropQueue(String clientID) {
-        LOG.info("Removing pending messages. CId= {}.", clientID);
-        this.m_db.delete(clientID);
-    }
-
-    @Override
-    public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) {
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("Moving inflight message to 2nd phase ack state. CId= {}, messageID = {}.", clientID, messageID);
-        }
-        Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
-            LOG.warn("Unable to retrieve inflight message record. CId= {}, messageId = {}.", clientID, messageID);
-            return;
-        }
-        MessageGUID guid = m.remove(messageID);
-
-        // remove from the ids store
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient != null) {
-            inFlightForClient.remove(messageID);
-        }
-
-        Map<Integer, MessageGUID> messageIDs = Utils
-                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
-        messageIDs.put(messageID, guid);
-        m_secondPhaseStore.put(clientID, messageIDs);
-    }
-
-    @Override
-    public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) {
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID);
-        }
-        Map<Integer, MessageGUID> messageIDs = Utils
-                .defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
-        MessageGUID guid = messageIDs.remove(messageID);
-        m_secondPhaseStore.put(clientID, messageIDs);
-        return guid;
-    }
-
-    @Override
-    public StoredMessage getInflightMessage(String clientID, int messageID) {
-        LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID);
-        Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientID);
-        if (clientEntries == null) {
-            LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID);
-            return null;
-        }
-        MessageGUID guid = clientEntries.get(messageID);
-        if (guid == null) {
-            LOG.warn("The message ID does not have an associated GUID. CId= {}, messageId = {}.", clientID, messageID);
-            return null;
-        }
-        return m_messagesStore.getMessageByGuid(guid);
-    }
-
-    @Override
-    public int getInflightMessagesNo(String clientID) {
-        if (!m_inflightStore.containsKey(clientID))
-            return 0;
-        else
-            return m_inflightStore.get(clientID).size();
-    }
-
-    @Override
-    public int getPendingPublishMessagesNo(String clientID) {
-        return m_messagesStore.getPendingPublishMessages(clientID);
-    }
-
-    @Override
-    public int getSecondPhaseAckPendingMessages(String clientID) {
-        if (!m_secondPhaseStore.containsKey(clientID))
-            return 0;
-        else
-            return m_secondPhaseStore.get(clientID).size();
-    }
-}