Minor fixes, added some javadoc and corrected a method name typo
This commit is contained in:
@@ -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<StoredMessage> searchMatching(IMatchingCondition condition);
|
||||
|
||||
void storePublishForFuture(PublishEvent evt);
|
||||
|
||||
List<PublishEvent> 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<PublishEvent> retrievePersistedPublishes(String clientID);
|
||||
|
||||
void cleanPersistedPublishMessage(String clientID, int messageID);
|
||||
|
||||
|
||||
+5
-2
@@ -134,6 +134,7 @@ public class HawtDBStorageService implements IStorageService {
|
||||
m_qos2Store = (SortedIndex<String, StoredPublishEvent>) 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<StoredMessage> 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<PublishEvent> retrivePersistedPublishes(String clientID) {
|
||||
@Override
|
||||
public List<PublishEvent> retrievePersistedPublishes(String clientID) {
|
||||
List<StoredPublishEvent> 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));
|
||||
}
|
||||
|
||||
|
||||
+2
-1
@@ -74,7 +74,8 @@ public class MemoryStorageService implements IStorageService {
|
||||
m_persistentMessageStore.put(clientID, storedEvents);
|
||||
}
|
||||
|
||||
public List<PublishEvent> retrivePersistedPublishes(String clientID) {
|
||||
@Override
|
||||
public List<PublishEvent> retrievePersistedPublishes(String clientID) {
|
||||
return m_persistentMessageStore.get(clientID);
|
||||
}
|
||||
|
||||
|
||||
+3
-3
@@ -176,7 +176,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
|
||||
|
||||
private void republishStored(String clientID) {
|
||||
LOG.trace("republishStored invoked");
|
||||
List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(clientID);
|
||||
List<PublishEvent> 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<ValueEvent> {
|
||||
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<ValueEvent> {
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
Reference in New Issue
Block a user