Separated ISessionsStore and IMessagesStore MapDB implementations

broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java

@@ -141,7 +141,7 @@ public interface IMessagesStore {
      * */
     int nextPacketID(String clientID);
 
-    void close();
+//    void close();
 
     void cleanRetained(String topic);
 }

broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java

@@ -28,6 +28,8 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
  */
 public interface ISessionsStore {
 
+    void initStore();
+
     /**
      * Add a new subscription to the session
      */

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -626,6 +626,7 @@ public class ProtocolProcessor {
         subscriptions.add(newSubscription);
 
         //scans retained messages to be published to the new subscription
+        //TODO this is ugly, it does a linear scan on potential big dataset
         Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
             public boolean match(String key) {
                 return SubscriptionsStore.matchTopics(key, topic);

broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java

@@ -48,9 +48,7 @@ public class SimpleMessaging {
 
     private SubscriptionsStore subscriptions;
 
-
+    private MapDBPersistentStore m_mapStorage;
-    private IMessagesStore m_storageService;
-    private ISessionsStore m_sessionsStore;
 
     private BrokerInterceptor m_interceptor;
 
@@ -72,11 +70,10 @@ public class SimpleMessaging {
         subscriptions = new SubscriptionsStore();
 
         //TODO use a property to select the storage path
-        MapDBPersistentStore mapStorage = new MapDBPersistentStore(props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, ""));
+        m_mapStorage = new MapDBPersistentStore(props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, ""));
-        m_storageService = mapStorage;
+        m_mapStorage.initStore();
-        m_sessionsStore = mapStorage;
+        IMessagesStore messagesStore = m_mapStorage.messagesStore();
-
+        ISessionsStore sessionsStore = m_mapStorage.sessionsStore(messagesStore);
-        m_storageService.initStore();
 
         List<InterceptHandler> observers = new ArrayList<>();
         String interceptorClassName = props.getProperty("intercept.handler");
@@ -90,7 +87,7 @@ public class SimpleMessaging {
         }
         m_interceptor = new BrokerInterceptor(observers);
 
-        subscriptions.init(m_sessionsStore);
+        subscriptions.init(sessionsStore);
 
         String configPath = System.getProperty("moquette.path", null);
         String authenticatorClassName = props.getProperty(Constants.AUTHENTICATOR_CLASS_NAME, "");
@@ -136,7 +133,7 @@ public class SimpleMessaging {
         }
 
         boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
-        m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor);
+        m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor);
         return m_processor;
     }
     
@@ -176,6 +173,6 @@ public class SimpleMessaging {
     }
 
     public void shutdown() {
-        this.m_storageService.close();
+        this.m_mapStorage.close();
     }
 }

broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBMessagesStore.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright (c) 2012-2015 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 org.eclipse.moquette.spi.persistence;
+
+import org.eclipse.moquette.spi.IMatchingCondition;
+import org.eclipse.moquette.spi.IMessagesStore;
+import org.mapdb.DB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.eclipse.moquette.spi.persistence.MapDBSessionsStore.messageId2GuidsMapName;
+
+/**
+ * IMessagesStore implementation backed by MapDB.
+ *
+ * @author andrea
+ */
+class MapDBMessagesStore implements IMessagesStore {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MapDBMessagesStore.class);
+
+    private DB m_db;
+
+    //map clientID <-> set of currently in flight packet identifiers
+    private Map<String, Set<Integer>> m_inFlightIds;
+    //maps clientID -> guid
+    private ConcurrentMap<String, String> m_retainedStore;
+    //maps guid to message, it's message store
+    private ConcurrentMap<String, IMessagesStore.StoredMessage> m_persistentMessageStore;
+
+
+    MapDBMessagesStore(DB db) {
+        m_db = db;
+    }
+
+    @Override
+    public void initStore() {
+        m_retainedStore = m_db.getHashMap("retained");
+        m_persistentMessageStore = m_db.getHashMap("persistedMessages");
+        m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
+    }
+
+    @Override
+    public void storeRetained(String topic, String 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<String, String> entry : m_retainedStore.entrySet()) {
+            final String guid = entry.getValue();
+            StoredMessage storedMsg = m_persistentMessageStore.get(guid);
+            if (condition.match(entry.getKey())) {
+                results.add(storedMsg);
+            }
+        }
+
+        return results;
+    }
+
+    @Override
+    public String storePublishForFuture(StoredMessage evt) {
+        LOG.debug("storePublishForFuture store evt {}", evt);
+        if (evt.getClientID() == null) {
+            LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt);
+            throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
+        }
+        String guid = UUID.randomUUID().toString();
+        evt.setGuid(guid);
+        m_persistentMessageStore.put(guid, evt);
+        ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(evt.getClientID()));
+        messageIdToGuid.put(evt.getMessageID(), guid);
+        return guid;
+    }
+
+    @Override
+    public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
+        List<StoredMessage> ret = new ArrayList<>();
+        for (String guid : guids) {
+            ret.add(m_persistentMessageStore.get(guid));
+        }
+        return ret;
+    }
+
+    @Override
+    public void dropMessagesInSession(String clientID) {
+        m_db.getHashMap(messageId2GuidsMapName(clientID)).clear();
+        m_persistentMessageStore.remove(clientID);
+    }
+
+    @Override
+    public StoredMessage getMessageByGuid(String guid) {
+        return m_persistentMessageStore.get(guid);
+    }
+
+    /**
+     * Return the next valid packetIdentifier for the given client session.
+     * */
+    @Override
+    public int nextPacketID(String clientID) {
+        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
+        if (inFlightForClient == null) {
+            int nextPacketId = 1;
+            inFlightForClient = new HashSet<>();
+            inFlightForClient.add(nextPacketId);
+            this.m_inFlightIds.put(clientID, inFlightForClient);
+            return nextPacketId;
+        }
+        int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
+        int nextPacketId = (maxId + 1) % 0xFFFF;
+        inFlightForClient.add(nextPacketId);
+        return nextPacketId;
+    }
+
+    @Override
+    public void cleanRetained(String topic) {
+        m_retainedStore.remove(topic);
+    }
+}

broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java

@@ -17,11 +17,8 @@
 package org.eclipse.moquette.spi.persistence;
 
 import org.eclipse.moquette.proto.MQTTException;
-import org.eclipse.moquette.spi.ClientSession;
-import org.eclipse.moquette.spi.IMatchingCondition;
 import org.eclipse.moquette.spi.IMessagesStore;
 import org.eclipse.moquette.spi.ISessionsStore;
-import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
 import org.mapdb.DB;
 import org.mapdb.DBMaker;
 import org.slf4j.Logger;
@@ -30,18 +27,30 @@ import org.slf4j.LoggerFactory;
 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
-import java.util.*;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
-
 /**
  * MapDB main persistence implementation
  */
-public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
+public class MapDBPersistentStore {
+
+    /**
+     * Factory method to create message store backed by MapDB
+     * */
+    public IMessagesStore messagesStore() {
+        //TODO check m_db is valid and
+        IMessagesStore msgStore = new MapDBMessagesStore(m_db);
+        msgStore.initStore();
+        return msgStore;
+    }
+
+    public ISessionsStore sessionsStore(IMessagesStore msgStore) {
+        ISessionsStore sessionsStore = new MapDBSessionsStore(m_db, msgStore);
+        sessionsStore.initStore();
+        return sessionsStore;
+    }
 
     /**
      * This is a DTO used to persist minimal status (clean session and activation status) of
@@ -59,63 +68,30 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
 
     private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
 
-    //maps clientID -> guid
-    private ConcurrentMap<String, String> m_retainedStore;
-    //maps guid to message, it's message store
-    private ConcurrentMap<String, StoredMessage> m_persistentMessageStore;
-    //maps clientID->[MessageId -> guid]
-    private ConcurrentMap<String, Map<Integer, String>> m_inflightStore;
-    //map clientID <-> set of currently in flight packet identifiers
-    Map<String, Set<Integer>> m_inFlightIds;
-    //persistent Map of clientID, set of Subscriptions
-    private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
-    private ConcurrentMap<String, PersistentSession> m_persistentSessions;
-
-    //maps clientID->[guid*], insertion order cares, it's queue
-    private ConcurrentMap<String, List<String>> m_enqueuedStore;
-    //maps clientID->[messageID*]
-    private ConcurrentMap<String, Set<Integer>> m_secondPhaseStore;
-
     private DB m_db;
     private String m_storePath;
 
     protected final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1);
 
-    /*
-     * The default constructor will create an in memory store as no file path was specified
-     */
-    public MapDBPersistentStore() {
-    	this.m_storePath = "";
-    }
-    
     public MapDBPersistentStore(String storePath) {
         this.m_storePath = storePath;
     }
     
-    @Override
     public void initStore() {
-    	if (m_storePath == null || m_storePath.isEmpty()) {
+        if (m_storePath == null || m_storePath.isEmpty()) {
-    		m_db = DBMaker.newMemoryDB().make();
+            m_db = DBMaker.newMemoryDB().make();
-    	} else {
+        } else {
-	        File tmpFile;
+            File tmpFile;
-	        try {
+            try {
-	            tmpFile = new File(m_storePath);
+                tmpFile = new File(m_storePath);
-	            boolean fileNewlyCreated = tmpFile.createNewFile();
+                boolean fileNewlyCreated = tmpFile.createNewFile();
                 LOG.info("Starting with {} [{}] db file", fileNewlyCreated ? "fresh" : "existing", m_storePath);
-	        } catch (IOException ex) {
+            } catch (IOException ex) {
-	            LOG.error(null, ex);
+                LOG.error(null, ex);
-	            throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex);
+                throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex);
-	        }
+            }
-	        m_db = DBMaker.newFileDB(tmpFile).make();
+            m_db = DBMaker.newFileDB(tmpFile).make();
-    	}
+        }
-        m_retainedStore = m_db.getHashMap("retained");
-        m_persistentMessageStore = m_db.getHashMap("persistedMessages");
-        m_inflightStore = m_db.getHashMap("inflight");
-        m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
-        m_persistentSubscriptions = m_db.getHashMap("subscriptions");
-        m_persistentSessions = m_db.getHashMap("sessions");
-        m_enqueuedStore = m_db.getHashMap("sessionQueue");
-        m_secondPhaseStore = m_db.getHashMap("secondPhase");
         m_scheduler.scheduleWithFixedDelay(new Runnable() {
             @Override
             public void run() {
@@ -124,287 +100,13 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
         }, 30, 30, TimeUnit.SECONDS);
     }
 
-    @Override
-    public void cleanRetained(String topic) {
-        m_retainedStore.remove(topic);
-    }
-
-    @Override
-    public void storeRetained(String topic, String 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<String, String> entry : m_retainedStore.entrySet()) {
-            final String guid = entry.getValue();
-            StoredMessage storedMsg = m_persistentMessageStore.get(guid);
-            if (condition.match(entry.getKey())) {
-                results.add(storedMsg);
-            }
-        }
-
-        return results;
-    }
-
-    @Override
-    public String storePublishForFuture(StoredMessage evt) {
-        LOG.debug("storePublishForFuture store evt {}", evt);
-        if (evt.getClientID() == null) {
-            LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt);
-            throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
-        }
-        String guid = UUID.randomUUID().toString();
-        evt.setGuid(guid);
-        m_persistentMessageStore.put(guid, evt);
-        ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(evt.getClientID()));
-        messageIdToGuid.put(evt.getMessageID(), guid);
-        return guid;
-    }
-
-    @Override
-    public List<StoredMessage> listMessagesInSession(Collection<String> guids) {
-        List<StoredMessage> ret = new ArrayList<>();
-        for (String guid : guids) {
-            ret.add(m_persistentMessageStore.get(guid));
-        }
-        return ret;
-    }
-
-    @Override
-    public void dropMessagesInSession(String clientID) {
-        m_db.getHashMap(messageId2GuidsMapName(clientID)).clear();
-        m_persistentMessageStore.remove(clientID);
-    }
-
-    @Override
-    public StoredMessage getMessageByGuid(String guid) {
-        return m_persistentMessageStore.get(guid);
-    }
-
-    @Override
-    public String mapToGuid(String clientID, int messageID) {
-        ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
-        return messageIdToGuid.get(messageID);
-    }
-
-    private String messageId2GuidsMapName(String clientID) {
-        return "guidsMapping_" + clientID;
-    }
-
-    /**
-     * Return the next valid packetIdentifier for the given client session.
-     * */
-    @Override
-    public int nextPacketID(String clientID) {
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient == null) {
-            int nextPacketId = 1;
-            inFlightForClient = new HashSet<>();
-            inFlightForClient.add(nextPacketId);
-            this.m_inFlightIds.put(clientID, inFlightForClient);
-            return nextPacketId;
-        }
-        int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
-        int nextPacketId = (maxId + 1) % 0xFFFF;
-        inFlightForClient.add(nextPacketId);
-        return nextPacketId;
-    }
-
-    @Override
-    public void removeSubscription(String topic, String clientID) {
-        LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
-        if (!m_persistentSubscriptions.containsKey(clientID)) {
-            return;
-        }
-        Set<Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
-        //search for the subscription to remove
-        Subscription toBeRemoved = null;
-        for (Subscription sub : clientSubscriptions) {
-            if (sub.getTopicFilter().equals(topic)) {
-                toBeRemoved = sub;
-                break;
-            }
-        }
-
-        if (toBeRemoved != null) {
-            clientSubscriptions.remove(toBeRemoved);
-        }
-        m_persistentSubscriptions.put(clientID, clientSubscriptions);
-        if (LOG.isTraceEnabled()) {
-            LOG.trace("persisted subscriptions {}", m_persistentSubscriptions);
-        }
-    }
-
-    @Override
-    public void addNewSubscription(Subscription newSubscription) {
-        LOG.debug("addNewSubscription invoked with subscription {}", newSubscription);
-        final String clientID = newSubscription.getClientId();
-        if (!m_persistentSubscriptions.containsKey(clientID)) {
-            LOG.debug("subscriptions for clientID <{}> not present, empty subscriptions set", clientID);
-            m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
-        }
-
-        Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
-        if (!subs.contains(newSubscription)) {
-            LOG.debug("updating <{}> subscriptions set with new subscription", clientID);
-            //TODO check the subs doesn't contain another subscription to the same topic with different
-            Subscription existingSubscription = null;
-            for (Subscription scanSub : subs) {
-                if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) {
-                    existingSubscription = scanSub;
-                    break;
-                }
-            }
-            if (existingSubscription != null) {
-                subs.remove(existingSubscription);
-            }
-            subs.add(newSubscription);
-            m_persistentSubscriptions.put(clientID, subs);
-            LOG.debug("clientID {} subscriptions set now is {}", clientID, subs);
-        }
-    }
-
-    @Override
-    public void wipeSubscriptions(String clientID) {
-        m_persistentSubscriptions.remove(clientID);
-    }
-
-    @Override
-    public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
-        m_persistentSubscriptions.put(clientID, subscriptions);
-    }
-
-    public List<Subscription> listAllSubscriptions() {
-        List<Subscription> allSubscriptions = new ArrayList<>();
-        for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
-            allSubscriptions.addAll(entry.getValue());
-        }
-        LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions);
-        return allSubscriptions;
-    }
-
-    @Override
-    public boolean contains(String clientID) {
-        return m_persistentSubscriptions.containsKey(clientID);
-    }
-
-    @Override
-    public ClientSession createNewSession(String clientID, boolean cleanSession) {
-        LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession);
-        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 HashSet<Subscription>());
-        m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false));
-        return new ClientSession(clientID, this, this, cleanSession);
-    }
-
-    @Override
-    public ClientSession sessionForClient(String clientID) {
-        if (!m_persistentSessions.containsKey(clientID)) {
-            return null;
-        }
-
-        PersistentSession storedSession = m_persistentSessions.get(clientID);
-        ClientSession clientSession = new ClientSession(clientID, this, this, storedSession.cleanSession);
-        if (storedSession.active) {
-            clientSession.activate();
-        }
-        return clientSession;
-    }
-
-    @Override
-    public void activate(String clientID) {
-        activationHelper(clientID, true);
-    }
-
-    @Override
-    public void deactivate(String clientID) {
-        activationHelper(clientID, false);
-    }
-
-    @Override
-    public void inFlightAck(String clientID, int messageID) {
-        Map<Integer, String> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
-            LOG.error("Can't find the inFlight record for client <{}>", clientID);
-            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, String guid) {
-        Map<Integer, String> m = this.m_inflightStore.get(clientID);
-        if (m == null) {
-            m = new HashMap<>();
-        }
-        m.put(messageID, guid);
-        this.m_inflightStore.put(clientID, m);
-    }
-
-    @Override
-    public void bindToDeliver(String guid, String clientID) {
-        List<String> guids = defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
-        guids.add(guid);
-        m_enqueuedStore.put(clientID, guids);
-    }
-
-    @Override
-    public Collection<String> enqueued(String clientID) {
-        return defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
-    }
-
-    @Override
-    public void removeEnqueued(String clientID, String guid) {
-        List<String> guids = defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
-        guids.remove(guid);
-        m_enqueuedStore.put(clientID, guids);
-    }
-
-    @Override
-    public void secondPhaseAcknowledged(String clientID, int messageID) {
-        Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
-        messageIDs.remove(messageID);
-        m_secondPhaseStore.put(clientID, messageIDs);
-    }
-
-    @Override
-    public void secondPhaseAckWaiting(String clientID, int messageID) {
-        Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
-        messageIDs.add(messageID);
-        m_secondPhaseStore.put(clientID, messageIDs);
-    }
-
-    private void activationHelper(String clientID, boolean activation) {
-        PersistentSession storedSession = m_persistentSessions.get(clientID);
-        if (storedSession == null) {
-            throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null);
-        }
-        PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation);
-        m_persistentSessions.put(clientID, newStoredSession);
-    }
-
-    @Override
     public void close() {
         if (this.m_db.isClosed()) {
             LOG.debug("already closed");
             return;
         }
         this.m_db.commit();
-        LOG.debug("persisted subscriptions {}", m_persistentSubscriptions);
+        //LOG.debug("persisted subscriptions {}", m_persistentSubscriptions);
         this.m_db.close();
         LOG.debug("closed disk storage");
         this.m_scheduler.shutdown();

broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBSessionsStore.java

@@ -0,0 +1,264 @@
+/*
+ * Copyright (c) 2012-2015 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 org.eclipse.moquette.spi.persistence;
+
+import org.eclipse.moquette.spi.ClientSession;
+import org.eclipse.moquette.spi.IMessagesStore;
+import org.eclipse.moquette.spi.ISessionsStore;
+import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
+import org.eclipse.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
+import org.mapdb.DB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
+
+/**
+ * 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, Map<Integer, String>> m_inflightStore;
+    //map clientID <-> set of currently in flight packet identifiers
+    private Map<String, Set<Integer>> m_inFlightIds;
+    //persistent Map of clientID, set of Subscriptions
+    private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
+    private ConcurrentMap<String, PersistentSession> m_persistentSessions;
+    //maps clientID->[guid*], insertion order cares, it's queue
+    private ConcurrentMap<String, List<String>> m_enqueuedStore;
+    //maps clientID->[messageID*]
+    private ConcurrentMap<String, Set<Integer>> 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_persistentSubscriptions = m_db.getHashMap("subscriptions");
+        m_persistentSessions = m_db.getHashMap("sessions");
+        m_enqueuedStore = m_db.getHashMap("sessionQueue");
+        m_secondPhaseStore = m_db.getHashMap("secondPhase");
+    }
+
+    @Override
+    public void addNewSubscription(Subscription newSubscription) {
+        LOG.debug("addNewSubscription invoked with subscription {}", newSubscription);
+        final String clientID = newSubscription.getClientId();
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
+            LOG.debug("subscriptions for clientID <{}> not present, empty subscriptions set", clientID);
+            m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
+        }
+
+        Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
+        if (!subs.contains(newSubscription)) {
+            LOG.debug("updating <{}> subscriptions set with new subscription", clientID);
+            //TODO check the subs doesn't contain another subscription to the same topic with different
+            Subscription existingSubscription = null;
+            for (Subscription scanSub : subs) {
+                if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) {
+                    existingSubscription = scanSub;
+                    break;
+                }
+            }
+            if (existingSubscription != null) {
+                subs.remove(existingSubscription);
+            }
+            subs.add(newSubscription);
+            m_persistentSubscriptions.put(clientID, subs);
+            LOG.debug("clientID {} subscriptions set now is {}", clientID, subs);
+        }
+    }
+
+    @Override
+    public void removeSubscription(String topic, String clientID) {
+        LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
+            return;
+        }
+        Set<Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
+        //search for the subscription to remove
+        Subscription toBeRemoved = null;
+        for (Subscription sub : clientSubscriptions) {
+            if (sub.getTopicFilter().equals(topic)) {
+                toBeRemoved = sub;
+                break;
+            }
+        }
+
+        if (toBeRemoved != null) {
+            clientSubscriptions.remove(toBeRemoved);
+        }
+        m_persistentSubscriptions.put(clientID, clientSubscriptions);
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("persisted subscriptions {}", m_persistentSubscriptions);
+        }
+    }
+
+    @Override
+    public void wipeSubscriptions(String clientID) {
+        m_persistentSubscriptions.remove(clientID);
+    }
+
+    @Override
+    public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
+        m_persistentSubscriptions.put(clientID, subscriptions);
+    }
+
+    @Override
+    public List<Subscription> listAllSubscriptions() {
+        List<Subscription> allSubscriptions = new ArrayList<>();
+        for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
+            allSubscriptions.addAll(entry.getValue());
+        }
+        LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions);
+        return allSubscriptions;
+    }
+
+    @Override
+    public boolean contains(String clientID) {
+        return m_persistentSubscriptions.containsKey(clientID);
+    }
+
+    @Override
+    public ClientSession createNewSession(String clientID, boolean cleanSession) {
+        LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession);
+        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 HashSet<Subscription>());
+        m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false));
+        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);
+        ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
+        if (storedSession.active) {
+            clientSession.activate();
+        }
+        return clientSession;
+    }
+
+    @Override
+    public void activate(String clientID) {
+        activationHelper(clientID, true);
+    }
+
+    @Override
+    public void deactivate(String clientID) {
+        activationHelper(clientID, false);
+    }
+
+    private void activationHelper(String clientID, boolean activation) {
+        PersistentSession storedSession = m_persistentSessions.get(clientID);
+        if (storedSession == null) {
+            throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null);
+        }
+        PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation);
+        m_persistentSessions.put(clientID, newStoredSession);
+    }
+
+    @Override
+    public void inFlightAck(String clientID, int messageID) {
+        Map<Integer, String> m = this.m_inflightStore.get(clientID);
+        if (m == null) {
+            LOG.error("Can't find the inFlight record for client <{}>", clientID);
+            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, String guid) {
+        Map<Integer, String> m = this.m_inflightStore.get(clientID);
+        if (m == null) {
+            m = new HashMap<>();
+        }
+        m.put(messageID, guid);
+        this.m_inflightStore.put(clientID, m);
+    }
+
+    @Override
+    public void bindToDeliver(String guid, String clientID) {
+        List<String> guids = defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
+        guids.add(guid);
+        m_enqueuedStore.put(clientID, guids);
+    }
+
+    @Override
+    public Collection<String> enqueued(String clientID) {
+        return defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
+    }
+
+    @Override
+    public void removeEnqueued(String clientID, String guid) {
+        List<String> guids = defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
+        guids.remove(guid);
+        m_enqueuedStore.put(clientID, guids);
+    }
+
+    @Override
+    public void secondPhaseAcknowledged(String clientID, int messageID) {
+        Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
+        messageIDs.remove(messageID);
+        m_secondPhaseStore.put(clientID, messageIDs);
+    }
+
+    @Override
+    public void secondPhaseAckWaiting(String clientID, int messageID) {
+        Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
+        messageIDs.add(messageID);
+        m_secondPhaseStore.put(clientID, messageIDs);
+    }
+
+    @Override
+    public String mapToGuid(String clientID, int messageID) {
+        ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
+        return messageIdToGuid.get(messageID);
+    }
+
+    static String messageId2GuidsMapName(String clientID) {
+        return "guidsMapping_" + clientID;
+    }
+}

broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java

@@ -136,11 +136,6 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
         return nextPacketId;
     }
 
-    @Override
-    public void close() {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
     @Override
     public void addNewSubscription(Subscription newSubscription) {
         m_sessionsStore.addNewSubscription(newSubscription);

broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java

@@ -17,6 +17,8 @@ package org.eclipse.moquette.spi.persistence;
 
 import org.eclipse.moquette.proto.messages.AbstractMessage;
 import org.eclipse.moquette.server.IntegrationUtils;
+import org.eclipse.moquette.spi.IMessagesStore;
+import org.eclipse.moquette.spi.ISessionsStore;
 import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
 import org.junit.After;
 import org.junit.Before;
@@ -35,12 +37,16 @@ import static org.junit.Assert.*;
 public class MapDBPersistentStoreTest {
 
     MapDBPersistentStore m_storageService;
+    ISessionsStore m_sessionsStore;
+    IMessagesStore m_messagesStore;
 
     @Before
     public void setUp() throws Exception {
         IntegrationUtils.cleanPersistenceFile(DEFAULT_PERSISTENT_PATH);
         m_storageService = new MapDBPersistentStore(DEFAULT_PERSISTENT_PATH);
         m_storageService.initStore();
+        m_messagesStore = m_storageService.messagesStore();
+        m_sessionsStore = m_storageService.sessionsStore(m_messagesStore);
     }
 
     @After
@@ -55,12 +61,12 @@ public class MapDBPersistentStoreTest {
     @Test
     public void overridingSubscriptions() {
         Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false);
-        m_storageService.addNewSubscription(oldSubscription);
+        m_sessionsStore.addNewSubscription(oldSubscription);
         Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE, false);
-        m_storageService.addNewSubscription(overrindingSubscription);
+        m_sessionsStore.addNewSubscription(overrindingSubscription);
         
         //Verify
-        List<Subscription> subscriptions = m_storageService.listAllSubscriptions();
+        List<Subscription> subscriptions = m_sessionsStore.listAllSubscriptions();
         assertEquals(1, subscriptions.size());
         Subscription sub = subscriptions.get(0);
         assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos());
@@ -68,18 +74,18 @@ public class MapDBPersistentStoreTest {
 
     @Test
     public void testNextPacketID_notExistingClientSession() {
-        int packetId = m_storageService.nextPacketID("NOT_EXISTING_CLI");
+        int packetId = m_messagesStore.nextPacketID("NOT_EXISTING_CLI");
         assertEquals(1, packetId);
     }
 
     @Test
     public void testNextPacketID_existingClientSession() {
         //Force creation of inflight map for the CLIENT session
-        int packetId = m_storageService.nextPacketID("CLIENT");
+        int packetId = m_messagesStore.nextPacketID("CLIENT");
         assertEquals(1, packetId);
 
         //request a second packetID
-        packetId = m_storageService.nextPacketID("CLIENT");
+        packetId = m_messagesStore.nextPacketID("CLIENT");
         assertEquals(2, packetId);
     }
 
@@ -87,15 +93,15 @@ public class MapDBPersistentStoreTest {
     public void testNextPacketID() {
         //request a first ID
 
-        int packetId = m_storageService.nextPacketID("CLIENT");
+        int packetId = m_messagesStore.nextPacketID("CLIENT");
-        m_storageService.inFlight("CLIENT", packetId, "ABCDE"); //simulate an inflight
+        m_sessionsStore.inFlight("CLIENT", packetId, "ABCDE"); //simulate an inflight
         assertEquals(1, packetId);
 
         //release the ID
-        m_storageService.inFlightAck("CLIENT", packetId);
+        m_sessionsStore.inFlightAck("CLIENT", packetId);
 
         //request a second packetID, counter restarts from 0
-        packetId = m_storageService.nextPacketID("CLIENT");
+        packetId = m_messagesStore.nextPacketID("CLIENT");
         assertEquals(1, packetId);
     }
 

broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java

@@ -73,6 +73,11 @@ public class MemorySessionStore implements ISessionsStore {
         }
     }
 
+    @Override
+    public void initStore() {
+
+    }
+
     @Override
     public void addNewSubscription(Subscription newSubscription) {
         final String clientID = newSubscription.getClientId();