Changed the remove method to work of a session instead of a clientID

ChangeLog.txt

@@ -1,4 +1,5 @@
 Version 0.18-SNAPSHOT:
+   [fix] Fixed SegmentedQueues not being cleaned up on session purge. (#833)
    [feature] Manage payload format indicator property, when set verify payload format. (#826)
    [refactoring] Refactory of PostOffice to pass publish message in hits entirety avoiding decomposition into single parameters. (#827)
    [feature] Add Netty native transport support on MacOS. Bundle all the native transport module by default (#806)

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

@@ -236,7 +236,7 @@ public SessionCreationResult(Session session, CreationModeEnum mode, boolean alr
     private void removeExpiredSession(ISessionsRepository.SessionData expiredSession) {
         final String expiredAt = expiredSession.expireAt().map(Instant::toString).orElse("UNDEFINED");
         LOG.debug("Removing session {}, expired on {}", expiredSession.clientId(), expiredAt);
-        remove(expiredSession.clientId());
+        remove(pool.get(expiredSession.clientId()));
         sessionsRepository.delete(expiredSession);
         subscriptionsDirectory.removeSharedSubscriptionsForClient(expiredSession.clientId());
     }
@@ -482,19 +482,19 @@ private void purgeSessionState(Session session) {
             throw new SessionCorruptedException("Session has already changed state: " + session);
         }
 
-        remove(session.getClientID());
+        remove(session);
         sessionsRepository.delete(session.getSessionData());
         subscriptionsDirectory.removeSharedSubscriptionsForClient(session.getClientID());
     }
 
-    void remove(String clientID) {
-        final Session old = pool.remove(clientID);
-        if (old != null) {
-            unsubscribe(old);
+    void remove(Session session) {
+        String clientID = session.getClientID();
+        if (pool.remove(clientID, session)) {
+            unsubscribe(session);
             // remove from expired tracker if present
             sessionExpirationService.untrack(clientID);
             loopsGroup.routeCommand(clientID, "Clean up removed session", () -> {
-                old.cleanUp();
+                session.cleanUp();
                 return null;
             });
         }