From 2ed87b2c1a52ac8e86011577faf2078bd34c9df6 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 16:18:09 +0200 Subject: [PATCH] Moved the part to manage Pusblish messages persistence to StorageService --- .../messaging/spi/IStorageService.java | 8 ++++ .../spi/impl/HawtDBStorageService.java | 38 +++++++++++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 25 ++---------- 3 files changed, 49 insertions(+), 22 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java index a2454eb2..6f0c9132 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -1,10 +1,12 @@ package org.dna.mqtt.moquette.messaging.spi; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; +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.SimpleMessaging.StoredMessage; import java.util.Collection; +import java.util.List; /** * Defines the SPI to be implemented by a StorageService that handle persistence of messages and subscriptions. @@ -20,4 +22,10 @@ public interface IStorageService { Collection searchMatching(IMatchingCondition condition); + void storePublishForFuture(PublishEvent evt); + + List retrivePersistedPublishes(String clientID); + + void cleanPersistedPublishes(String clientID); + } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 14a4654e..6491d64b 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -3,6 +3,8 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.fusesource.hawtbuf.codec.StringCodec; import org.fusesource.hawtdb.api.BTreeIndexFactory; @@ -30,6 +32,8 @@ public class HawtDBStorageService implements IStorageService { private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; + //maps clientID to the list of pending messages stored + private SortedIndex> m_persistentMessageStore; private SortedIndex m_retainedStore; public HawtDBStorageService(MultiIndexFactory multiIndexFactory) { @@ -38,6 +42,12 @@ public class HawtDBStorageService implements IStorageService { public void initStore() { + initRetainedStore(); + //init the message store for QoS 1/2 messages in clean sessions + initPersistentMessageStore(); + } + + private void initRetainedStore() { BTreeIndexFactory indexFactory = new BTreeIndexFactory(); indexFactory.setKeyCodec(StringCodec.INSTANCE); @@ -45,6 +55,14 @@ public class HawtDBStorageService implements IStorageService { } + + private void initPersistentMessageStore() { + BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); + } + public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { if (message.length == 0) { //clean the message from topic @@ -69,4 +87,24 @@ public class HawtDBStorageService implements IStorageService { return results; } + + public void storePublishForFuture(PublishEvent evt) { + List storedEvents; + String clientID = evt.getClientID(); + if (!m_persistentMessageStore.containsKey(clientID)) { + storedEvents = new ArrayList(); + } else { + storedEvents = m_persistentMessageStore.get(clientID); + } + storedEvents.add(evt); + m_persistentMessageStore.put(clientID, storedEvents); + } + + public List retrivePersistedPublishes(String clientID) { + return m_persistentMessageStore.get(clientID); + } + + public void cleanPersistedPublishes(String clientID) { + m_persistentMessageStore.remove(clientID); + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index c3ac3f53..2602dbf0 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -18,7 +18,6 @@ import org.dna.mqtt.moquette.messaging.spi.INotifier; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; -import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.Constants; import org.dna.mqtt.moquette.server.Server; @@ -70,8 +69,6 @@ public class SimpleMessaging implements IMessaging, Runnable { //bind clientID+MsgID -> evt message published private SortedIndex m_inflightStore; - //maps clientID to the list of pending messages stored - private SortedIndex> m_persistentMessageStore; private IStorageService m_storageService; @@ -94,8 +91,6 @@ public class SimpleMessaging implements IMessaging, Runnable { subscriptions.init(m_multiIndexFactory); initInflightMessageStore(); - //init the message store for QoS 1/2 messages in clean sessions - initPersistentMessageStore(); m_storageService = new HawtDBStorageService(m_multiIndexFactory); m_storageService.initStore(); @@ -113,12 +108,6 @@ public class SimpleMessaging implements IMessaging, Runnable { m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); } - private void initPersistentMessageStore() { - BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); - } public void setNotifier(INotifier notifier) { m_notifier= notifier; @@ -297,15 +286,7 @@ public class SimpleMessaging implements IMessaging, Runnable { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it if (!sub.isCleanSession() && !sub.isActive()) { - List storedEvents; - String clientID = evt.getClientID(); - if (!m_persistentMessageStore.containsKey(clientID)) { - storedEvents = new ArrayList(); - } else { - storedEvents = m_persistentMessageStore.get(clientID); - } - storedEvents.add(evt); - m_persistentMessageStore.put(clientID, storedEvents); + m_storageService.storePublishForFuture(evt); } m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); } @@ -354,7 +335,7 @@ public class SimpleMessaging implements IMessaging, Runnable { subscriptions.removeForClient(evt.getClientID()); //remove also the messages stored of type QoS1/2 - m_persistentMessageStore.remove(evt.getClientID()); + m_storageService.cleanPersistedPublishes(evt.getClientID()); } private void processDisconnect(DisconnectEvent evt) throws InterruptedException { @@ -375,7 +356,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } private void processRepublish(RepublishEvent evt) throws InterruptedException { - List publishedEvents = m_persistentMessageStore.get(evt.getClientID()); + List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); if (publishedEvents == null) { return; }