diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index a619bbd4..d9bb8c73 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/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 getSubscriptions() { + if (m_processorBootstrapper == null) { + return null; + } + return m_processorBootstrapper.getSubscriptions(); + } + /** * SPI method used by Broker embedded applications to add intercept handlers. * */ diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index e1237b01..6ff55a64 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/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 getSubscriptions(); /** * @return true iff there are subscriptions persisted with clientID diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 480fa596..55bfe981 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/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(messagesStore); + m_sessionsStore = m_mapStorage.sessionsStore(messagesStore); List 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 getSubscriptions() { + return m_sessionsStore.getSubscriptions(); + } + public void shutdown() { this.m_mapStorage.close(); } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index 0d7ead81..3204f010 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/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 getSubscriptions() { + List subscriptions = new ArrayList<>(); + for (String clientID : m_persistentSessions.keySet()) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); + subscriptions.addAll(clientSubscriptions.values()); + } + return subscriptions; + } + @Override public boolean contains(String clientID) { return m_db.exists("subscriptions_" + clientID); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index d17fdfd8..38038a6b 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/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(); + } } diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index 937fd310..d4d9d7ae 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -156,6 +156,15 @@ public class MemorySessionStore implements ISessionsStore { return null; } + @Override + public List getSubscriptions() { + List subscriptions = new ArrayList<>(); + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + subscriptions.addAll(entry.getValue()); + } + return subscriptions; + } + @Override public void inFlightAck(String clientID, int messageID) { Map m = this.m_inflightStore.get(clientID);