Added tracking of closed session time and expired session cleaner for clients that never reconnects but had clean session flag = false

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

@@ -25,10 +25,9 @@ import io.moquette.spi.impl.subscriptions.Topic;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.LocalDateTime;
 import java.util.*;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.*;
 
 public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
 
@@ -51,6 +50,7 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
     }
 
     private final Map<String, Session> sessions = new ConcurrentHashMap<>();
+    private ConcurrentNavigableMap<LocalDateTime, Set<String>> sessionsClosingTimes = new ConcurrentSkipListMap<>();
 
     MemorySessionStore() {
     }
@@ -280,4 +280,24 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore {
         // TODO this missing last step breaks the junit test
         //sessions.remove(clientID);
     }
+
+    @Override
+    public synchronized void trackSessionClose(LocalDateTime when, String clientID) {
+        this.sessionsClosingTimes.putIfAbsent(when, new HashSet<>());
+        this.sessionsClosingTimes.computeIfPresent(when, (key, oldSet) -> {
+            oldSet.add(clientID);
+            return oldSet;
+        });
+    }
+
+    @Override
+    public Set<String> sessionOlderThan(LocalDateTime queryPin) {
+        final Set<String> results = new HashSet<>();
+        LocalDateTime keyBefore = this.sessionsClosingTimes.lowerKey(queryPin);
+        while (keyBefore != null) {
+            results.addAll(this.sessionsClosingTimes.get(keyBefore));
+            keyBefore = this.sessionsClosingTimes.lowerKey(keyBefore);
+        }
+        return results;
+    }
 }

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

@@ -28,6 +28,8 @@ public class MemoryStorageService implements IStore {
     private MemorySessionStore m_sessionsStore;
     private MemoryMessagesStore m_messagesStore;
 
+    // NB these params must be here "by contract" used in introspection instantiation used in
+    // ProtocolProcessorBootstrapper.instantiateConfiguredStore
     public MemoryStorageService(IConfig props, ScheduledExecutorService scheduler) {
         m_messagesStore = new MemoryMessagesStore();
         m_sessionsStore = new MemorySessionStore();

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

@@ -19,8 +19,10 @@ package io.moquette.spi;
 import io.moquette.persistence.PersistentSession;
 import io.moquette.spi.IMessagesStore.StoredMessage;
 
+import java.time.LocalDateTime;
 import java.util.Collection;
 import java.util.Queue;
+import java.util.Set;
 
 /**
  * Store used to handle the persistence of the subscriptions tree.
@@ -115,4 +117,14 @@ public interface ISessionsStore {
     int countPubReleaseWaitingPubComplete(String clientID);
 
     void removeTemporaryQoS2(String clientID);
+
+    /**
+     * Tracks the closing time of a persistent session
+     * */
+    void trackSessionClose(LocalDateTime when, String clientID);
+
+    /**
+     * List the sessions ids closed before the pin date.
+     * */
+    Set<String> sessionOlderThan(LocalDateTime queryPin);
 }

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

@@ -97,7 +97,8 @@ public class ProtocolProcessorBootstrapper {
         store.initStore();
         messagesStore = store.messagesStore();
         m_sessionsStore = store.sessionsStore();
-        SessionsRepository sessionsRepository = new SessionsRepository(this.m_sessionsStore);
+        SessionsRepository sessionsRepository = new SessionsRepository(this.m_sessionsStore, server.getScheduler());
+        sessionsRepository.init();
         this.subscriptionsStore = m_sessionsStore.subscriptionStore();
         storeShutdown = new Runnable() {
 

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

@@ -13,7 +13,6 @@
  *
  * You may elect to redistribute this code under either of these licenses.
  */
-
 package io.moquette.spi.impl;
 
 import io.moquette.persistence.PersistentSession;
@@ -21,21 +20,44 @@ import io.moquette.spi.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public class SessionsRepository {
 
+    /**
+     * Task to be scheduled to execute the cleaning of persisted sessions (clean flag=false) older than a defined period.
+     */
+    private class SessionCleanerTask implements Runnable {
+
+        @Override
+        public void run() {
+            wipeExpiredSessions();
+        }
+    }
+
     private static final Logger LOG = LoggerFactory.getLogger(SessionsRepository.class);
     private final ISessionsStore sessions;
     private ISubscriptionsStore subscriptionsStore;
+    private ScheduledExecutorService scheduler;
     private final Map<String, ClientSession> sessionsCache = new ConcurrentHashMap<>();
 
-    public SessionsRepository(ISessionsStore sessionsStore) {
+    public SessionsRepository(ISessionsStore sessionsStore, ScheduledExecutorService scheduler) {
         this.sessions = sessionsStore;
         this.subscriptionsStore = sessionsStore.subscriptionStore();
+        this.scheduler = scheduler;
+    }
+
+    public void init() {
+        SessionCleanerTask cleanerTask = new SessionCleanerTask();
+        this.scheduler.schedule(cleanerTask, 1, TimeUnit.HOURS);
     }
 
     public ClientSession sessionForClient(String clientID) {
@@ -122,5 +144,15 @@ public class SessionsRepository {
         }
 
         sessionsCache.remove(clientId);
+        this.sessions.trackSessionClose(LocalDateTime.now(), clientId);
+    }
+
+    private void wipeExpiredSessions() {
+        final LocalDateTime pin = LocalDateTime.now().minus(6, ChronoUnit.DAYS);
+        final Set<String> expiredSessionsIds = this.sessions.sessionOlderThan(pin);
+        for (String expiredSession : expiredSessionsIds) {
+            this.sessions.removeDurableSession(expiredSession);
+            this.subscriptionsStore.wipeSubscriptions(expiredSession);
+        }
     }
 }