From b8a15d9a5d827dd234599338717b58543756cfcd Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Tue, 20 May 2014 23:02:52 +0200 Subject: [PATCH] Minor fixes, added some javadoc and corrected a method name typo --- .../moquette/messaging/spi/IStorageService.java | 14 +++++++++++++- .../messaging/spi/impl/HawtDBStorageService.java | 7 +++++-- .../messaging/spi/impl/MemoryStorageService.java | 3 ++- .../messaging/spi/impl/ProtocolProcessor.java | 6 +++--- 4 files changed, 23 insertions(+), 7 deletions(-) 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 f6c5bfd3..a8b6d4de 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 @@ -18,13 +18,25 @@ public interface IStorageService extends IPersistentSubscriptionStore { * */ void initStore(); + /** + * Persist the message. + * If the message is empty then the topic is cleaned, else it's stored. + */ void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos); + /** + * Return a list of retained messages that satisfy the condition. + */ Collection searchMatching(IMatchingCondition condition); void storePublishForFuture(PublishEvent evt); - List retrivePersistedPublishes(String clientID); + /** + * Return the list of persisted publishes for the given clientID. + * For QoS1 and QoS2 with clean session flag, this method return the list of + * missed publish events while the client was disconnected. + */ + List retrievePersistedPublishes(String clientID); void cleanPersistedPublishMessage(String clientID, int messageID); 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 ae60785e..1ac3f1c3 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 @@ -134,6 +134,7 @@ public class HawtDBStorageService implements IStorageService { m_qos2Store = (SortedIndex) m_multiIndexFactory.openOrCreate("qos2Store", indexFactory); } + @Override public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { if (!message.hasRemaining()) { //clean the message from topic @@ -146,6 +147,7 @@ public class HawtDBStorageService implements IStorageService { } } + @Override public Collection searchMatching(IMatchingCondition condition) { LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); @@ -175,7 +177,8 @@ public class HawtDBStorageService implements IStorageService { LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic()); } - public List retrivePersistedPublishes(String clientID) { + @Override + public List retrievePersistedPublishes(String clientID) { List storedEvts = m_persistentMessageStore.get(clientID); if (storedEvts == null) { return null; @@ -266,7 +269,7 @@ public class HawtDBStorageService implements IStorageService { /*-------- QoS 2 storage management --------------*/ public void persistQoS2Message(String publishKey, PublishEvent evt) { - LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt); + LOG.debug("persistQoS2Message store pubKey: {}, evt: {}", publishKey, evt); m_qos2Store.put(publishKey, convertToStored(evt)); } 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 79ca189e..f3eb3a6e 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 @@ -74,7 +74,8 @@ public class MemoryStorageService implements IStorageService { m_persistentMessageStore.put(clientID, storedEvents); } - public List retrivePersistedPublishes(String clientID) { + @Override + public List retrievePersistedPublishes(String clientID) { return m_persistentMessageStore.get(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 c4613d35..f62184a7 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 @@ -176,7 +176,7 @@ class ProtocolProcessor implements EventHandler { private void republishStored(String clientID) { LOG.trace("republishStored invoked"); - List publishedEvents = m_storageService.retrivePersistedPublishes(clientID); + List publishedEvents = m_storageService.retrievePersistedPublishes(clientID); if (publishedEvents == null) { LOG.info("No stored messages for client <{}>", clientID); return; @@ -224,7 +224,7 @@ class ProtocolProcessor implements EventHandler { sendPubRec(evt.getClientID(), evt.getMessageID()); } - //NB publish 2 subscribers for QoS 2 happen upon PUBREL from publsher + //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) { publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); } @@ -294,7 +294,7 @@ class ProtocolProcessor implements EventHandler { } private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) { - LOG.debug("sendPublish invoked clientId <{}> on topic <{}> QoS {} ratained {} messageID {}", clientId, topic, qos, retained, messageID); + LOG.debug("sendPublish invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID); PublishMessage pubMessage = new PublishMessage(); pubMessage.setRetainFlag(retained); pubMessage.setTopicName(topic);