Add simple connection management capabilities to the broker. An external codebase will now be able to monitor and close the opened connections.

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

@@ -189,4 +189,21 @@ public class ClientSession {
     public IMessagesStore.StoredMessage getInflightMessage(int messageID) {
         return m_sessionsStore.getInflightMessage(clientID, messageID);
     }
+
+    public Set<Subscription> getSubscriptions() {
+        return subscriptions;
+    }
+
+    public int getPendingPublishMessagesNo() {
+        return m_sessionsStore.getPendingPublishMessagesNo(clientID);
+    }
+
+    public int getSecondPhaseAckPendingMessages() {
+        return m_sessionsStore.getSecondPhaseAckPendingMessages(clientID);
+    }
+
+    public int getInflightMessagesNo() {
+        return m_sessionsStore.getInflightMessagesNo(clientID);
+    }
+
 }

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

@@ -135,4 +135,6 @@ public interface IMessagesStore {
     StoredMessage getMessageByGuid(MessageGUID guid);
 
     void cleanRetained(String topic);
+
+    int getPendingPublishMessages(String clientID);
 }

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

@@ -15,6 +15,7 @@
  */
 package io.moquette.spi;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 
@@ -117,6 +118,12 @@ public interface ISessionsStore {
      * @return the session for the given clientID, null if not found.
      * */
     ClientSession sessionForClient(String clientID);
+    
+    /**
+     * Returns all the sessions
+     * @return
+     */
+    Collection<ClientSession> getAllSessions();
 
     void inFlightAck(String clientID, int messageID);
 
@@ -156,4 +163,29 @@ public interface ISessionsStore {
     MessageGUID mapToGuid(String clientID, int messageID);
     
     StoredMessage getInflightMessage(String clientID, int messageID);
+
+    /**
+     * Returns the number of inflight messages for the given client ID
+     * 
+     * @param clientID
+     * @return
+     */
+    int getInflightMessagesNo(String clientID);
+
+    /**
+     * Returns the size of the session queue for the given client ID
+     * 
+     * @param clientID
+     * @return
+     */
+    int getPendingPublishMessagesNo(String clientID);
+
+    /**
+     * Returns the number of second-phase ACK pending messages for the given
+     * client ID
+     * 
+     * @param clientID
+     * @return
+     */
+    int getSecondPhaseAckPendingMessages(String clientID);
 }

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

@@ -164,7 +164,7 @@ public class ProtocolProcessor {
             IAuthenticator authenticator,
             boolean allowAnonymous,
             boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
-		init(new ConnectionDescriptorStore(), subscriptions, storageService, sessionsStore, authenticator,
+		init(new ConnectionDescriptorStore(sessionsStore), subscriptions, storageService, sessionsStore, authenticator,
 				allowAnonymous, allowZeroByteClientId, authorizator, interceptor, serverPort);
 	}
 

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

@@ -17,6 +17,7 @@ package io.moquette.spi.impl;
 
 import io.moquette.BrokerConstants;
 import io.moquette.interception.InterceptHandler;
+import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.server.Server;
 import io.moquette.server.config.IConfig;
 import io.moquette.server.config.IResourceLoader;
@@ -59,6 +60,8 @@ public class ProtocolProcessorBootstrapper {
     private BrokerInterceptor m_interceptor;
 
     private final ProtocolProcessor m_processor = new ProtocolProcessor();
+    
+    private ConnectionDescriptorStore connectionDescriptors;
 
     public ProtocolProcessorBootstrapper() {
     }
@@ -141,11 +144,14 @@ public class ProtocolProcessorBootstrapper {
             }
             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(subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
+        m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
         return m_processor;
     }
     
@@ -214,4 +220,8 @@ public class ProtocolProcessorBootstrapper {
     public void shutdown() {
         this.m_mapStorage.close();
     }
+
+    public ConnectionDescriptorStore getConnectionDescriptors() {
+        return connectionDescriptors;
+    }
 }

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

@@ -60,6 +60,10 @@ public final class Subscription implements Serializable {
         return topicFilter;
     }
 
+    public boolean isActive() {
+        return active;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

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

@@ -137,4 +137,11 @@ class MapDBMessagesStore implements IMessagesStore {
         }
         m_retainedStore.remove(topic);
     }
+
+    @Override
+    public int getPendingPublishMessages(String clientID) {
+        ConcurrentMap<Integer, MessageGUID> messageIdToGuidMap = m_db
+                .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
+        return messageIdToGuidMap.size();
+    }
 }

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

@@ -168,6 +168,15 @@ class MapDBSessionsStore implements ISessionsStore {
         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<ClientSession>();
+        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) {
@@ -318,4 +327,25 @@ class MapDBSessionsStore implements ISessionsStore {
         }
 		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();
+    }
 }