diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java index 17404dc4..49a32487 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java @@ -25,9 +25,15 @@ import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; */ public interface IPersistentSubscriptionStore { + /** + * Add a new subscription to the session + */ void addNewSubscription(Subscription newSubscription, String clientID); - void removeAllSubscriptions(String clientID); + /** + * Remove all the subscriptions of the session + */ + void wipeSubscriptions(String sessionID); List retrieveAllSubscriptions(); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index ce8e0900..979180fd 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -260,7 +260,7 @@ public class HawtDBStorageService implements IStorageService { } } - public void removeAllSubscriptions(String clientID) { + public void wipeSubscriptions(String clientID) { m_persistentSubscriptions.remove(clientID); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index 13b5d902..06e0af64 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -130,7 +130,7 @@ public class MemoryStorageService implements IStorageService { } } - public void removeAllSubscriptions(String clientID) { + public void wipeSubscriptions(String clientID) { m_persistentSubscriptions.remove(clientID); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 81db4d18..8b072410 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -160,7 +160,7 @@ class ProtocolProcessor implements EventHandler { boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION); if (cleanSession) { //cleanup topic subscriptions - processRemoveAllSubscriptions(msg.getClientID()); + cleanSession(msg.getClientID()); } m_clientIDs.get(msg.getClientID()).getSession().close(false); @@ -208,7 +208,7 @@ class ProtocolProcessor implements EventHandler { if (msg.isCleanSession()) { //remove all prev subscriptions //cleanup topic subscriptions - processRemoveAllSubscriptions(msg.getClientID()); + cleanSession(msg.getClientID()); } ConnAckMessage okResp = new ConnAckMessage(); @@ -243,10 +243,10 @@ class ProtocolProcessor implements EventHandler { m_storageService.cleanPersistedPublishMessage(clientID, messageID); } - private void processRemoveAllSubscriptions(String clientID) { + private void cleanSession(String clientID) { LOG.info("cleaning old saved subscriptions for client <{}>", clientID); //remove from log all subscriptions - m_storageService.removeAllSubscriptions(clientID); + m_storageService.wipeSubscriptions(clientID); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -458,7 +458,7 @@ class ProtocolProcessor implements EventHandler { void processDisconnect(ServerChannel session, String clientID, boolean cleanSession) throws InterruptedException { if (cleanSession) { //cleanup topic subscriptions - processRemoveAllSubscriptions(clientID); + cleanSession(clientID); } // m_notifier.disconnect(evt.getSession()); m_clientIDs.remove(clientID);