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 new file mode 100644 index 00000000..f27295d4 --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java @@ -0,0 +1,18 @@ +package org.dna.mqtt.moquette.messaging.spi; + +import java.util.List; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; + +/** + * Store used to handle the persistence of the subscriptions tree. + * + * @author andrea + */ +public interface IPersistentSubscriptionStore { + + void addNewSubscription(Subscription newSubscription, String clientID); + + void removeAllSubscriptions(String clientID); + + List retrieveAllSubscriptions(); +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java index 39e7c493..f6c5bfd3 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -1,7 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi; import java.nio.ByteBuffer; -import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import static org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage; @@ -12,7 +11,7 @@ import java.util.List; /** * Defines the SPI to be implemented by a StorageService that handle persistence of messages and subscriptions. */ -public interface IStorageService { +public interface IStorageService extends IPersistentSubscriptionStore { /** * Used to initialize all persistent store structures @@ -35,12 +34,6 @@ public interface IStorageService { void addInFlight(PublishEvent evt, String publishKey); - void addNewSubscription(Subscription newSubscription, String clientID); - - void removeAllSubscriptions(String clientID); - - List retrieveAllSubscriptions(); - void close(); void persistQoS2Message(String publishKey, PublishEvent evt); 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 c93396e1..1aef5991 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 @@ -60,8 +60,13 @@ class ProtocolProcessor implements EventHandler { ProtocolProcessor() {} - void init(/*Map clientIDs, */SubscriptionsStore subscriptions, - IStorageService storageService) { + /** + * @param subscriptions the subscription store where are stored all the existing + * clients subscriptions. + * @param storageService the persistent store to use for save/load of messages + * for QoS1 and QoS2 handling. + */ + void init(SubscriptionsStore subscriptions, IStorageService storageService) { //m_clientIDs = clientIDs; this.subscriptions = subscriptions; m_storageService = storageService; diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index c92ce9f7..7948e66f 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -6,8 +6,8 @@ import java.util.Collections; import java.util.List; import java.util.Queue; import java.util.concurrent.LinkedBlockingDeque; +import org.dna.mqtt.moquette.messaging.spi.IPersistentSubscriptionStore; -import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,13 +58,13 @@ public class SubscriptionsStore { private TreeNode subscriptions = new TreeNode(null); private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); - private IStorageService m_storageService; + private IPersistentSubscriptionStore m_storageService; /** * Initialize basic store structures, like the FS storage to maintain * client's topics subscriptions */ - public void init(IStorageService storageService) { + public void init(IPersistentSubscriptionStore storageService) { LOG.debug("init invoked"); m_storageService = storageService;