Merge pull request #212 from teranisi/get_subscriptions_api

Add 'getSubscriptions' API on Server.

broker/src/main/java/io/moquette/server/Server.java

@@ -33,6 +33,7 @@ import io.moquette.server.config.FilesystemConfig;
 import io.moquette.server.config.IConfig;
 import io.moquette.server.netty.NettyAcceptor;
 import io.moquette.spi.impl.ProtocolProcessor;
+import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.security.IAuthenticator;
 import io.moquette.spi.security.IAuthorizator;
 import io.moquette.spi.security.ISslContextCreator;
@@ -211,6 +212,17 @@ public class Server {
         LOG.info("Server stopped");
     }
 
+    /**
+     * SPI method used by Broker embedded applications to get list of subscribers.
+     * Returns null if the broker is not started.
+     */
+    public List<Subscription> getSubscriptions() {
+        if (m_processorBootstrapper == null) {
+            return null;
+        }
+        return m_processorBootstrapper.getSubscriptions();
+    }
+
     /**
      * SPI method used by Broker embedded applications to add intercept handlers.
      * */

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

@@ -93,6 +93,11 @@ public interface ISessionsStore {
      * @return the subscription stored by clientID and topicFilter, if any else null;
      * */
     Subscription getSubscription(ClientTopicCouple couple);
+    
+    /*
+     * @return all subscriptions stored.
+     */
+    List<Subscription> getSubscriptions();
 
     /**
      * @return true iff there are subscriptions persisted with clientID

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

@@ -22,6 +22,7 @@ import io.moquette.interception.InterceptHandler;
 import io.moquette.server.config.IConfig;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.security.*;
+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;
@@ -49,6 +50,8 @@ public class ProtocolProcessorBootstrapper {
     private SubscriptionsStore subscriptions;
 
     private MapDBPersistentStore m_mapStorage;
+    
+    private ISessionsStore m_sessionsStore;
 
     private BrokerInterceptor m_interceptor;
 
@@ -75,7 +78,7 @@ public class ProtocolProcessorBootstrapper {
         m_mapStorage = new MapDBPersistentStore(props);
         m_mapStorage.initStore();
         IMessagesStore messagesStore = m_mapStorage.messagesStore();
-        ISessionsStore sessionsStore = m_mapStorage.sessionsStore();
+        m_sessionsStore = m_mapStorage.sessionsStore();
 
         List<InterceptHandler> observers = new ArrayList<>(embeddedObservers);
         String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
@@ -95,7 +98,7 @@ public class ProtocolProcessorBootstrapper {
         }
         m_interceptor = new BrokerInterceptor(observers);
 
-        subscriptions.init(sessionsStore);
+        subscriptions.init(m_sessionsStore);
 
         String configPath = System.getProperty("moquette.path", null);
         String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "");
@@ -140,7 +143,7 @@ public class ProtocolProcessorBootstrapper {
 
         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, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
+        m_processor.init(subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
         return m_processor;
     }
     
@@ -191,6 +194,10 @@ public class ProtocolProcessorBootstrapper {
         return instance;
     }
 
+    public List<Subscription> getSubscriptions() {
+        return m_sessionsStore.getSubscriptions();
+    }
+
     public void shutdown() {
         this.m_mapStorage.close();
     }

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

@@ -26,17 +26,18 @@ import io.moquette.spi.ISessionsStore.ClientTopicCouple;
  * 
  * @author andrea
  */
-public class Subscription implements Serializable {
+public final class Subscription implements Serializable {
     
-    QOSType requestedQos; //max QoS acceptable
-    String clientId;
-    String topicFilter;
-    boolean active = true;
+    final QOSType requestedQos; //max QoS acceptable
+    final String clientId;
+    final String topicFilter;
+    final boolean active;
     
     public Subscription(String clientId, String topicFilter, QOSType requestedQos) {
         this.requestedQos = requestedQos;
         this.clientId = clientId;
         this.topicFilter = topicFilter;
+        this.active = true;
     }
 
     public Subscription(Subscription orig) {

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

@@ -117,6 +117,16 @@ class MapDBSessionsStore implements ISessionsStore {
         return clientSubscriptions.get(couple.topicFilter);
     }
 
+    @Override
+    public List<Subscription> getSubscriptions() {
+        List<Subscription> subscriptions = new ArrayList<>();
+        for (String clientID : m_persistentSessions.keySet()) {
+            ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
+            subscriptions.addAll(clientSubscriptions.values());
+        }
+        return subscriptions;
+    }
+
     @Override
     public boolean contains(String clientID) {
         return m_db.exists("subscriptions_" + clientID);

broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java

@@ -97,6 +97,17 @@ public class ServerIntegrationEmbeddedPublishTest {
         m_server.internalPublish(message);
     }
 
+    private void connectNoCleanSession() throws Exception {
+        MqttConnectOptions opts = new MqttConnectOptions();
+        opts.setCleanSession(false);
+        m_subscriber.connect(opts);
+    }
+
+    private void subscribeToWithQosAndNoCleanSession(String topic, int qos) throws Exception {
+        connectNoCleanSession();
+        m_subscriber.subscribe(topic, qos);
+    }
+
     private void verifyNoMessageIsReceived() throws Exception {
         MqttMessage msg = m_callback.getMessage(false);
         assertNull(msg);
@@ -276,4 +287,16 @@ public class ServerIntegrationEmbeddedPublishTest {
         MqttMessage message = m_callback.getMessage(true);
         assertTrue(message == null);
     }
+
+    @Test
+    public void testClientSubscribeWithoutCleanSession() throws Exception {
+        LOG.info("*** testClientSubscribeWithoutCleanSession ***");
+        subscribeToWithQosAndNoCleanSession("foo", 1);
+        m_subscriber.disconnect();
+        assertTrue(m_server.getSubscriptions().size() == 1);
+        connectNoCleanSession();
+        assertTrue(m_server.getSubscriptions().size() == 1);
+        internalPublishToWithQosAndRetained("foo", QOSType.MOST_ONE, false);
+        verifyMessageIsReceivedSuccessfully();
+    }
 }

broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java

@@ -157,6 +157,15 @@ public class MemorySessionStore implements ISessionsStore {
         return null;
     }
 
+    @Override
+    public List<Subscription> getSubscriptions() {
+        List<Subscription> subscriptions = new ArrayList<>();
+        for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
+            subscriptions.addAll(entry.getValue());
+        }
+        return subscriptions;
+    }
+
     @Override
     public void inFlightAck(String clientID, int messageID) {
         Map<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);