From 63333b11ad660445600c86ab3a74e5328516964f Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 18:08:06 +0200 Subject: [PATCH] Completed the memory implementation of DummyStorageService --- .../spi/impl/DummyStorageService.java | 55 +++++++++++++++---- .../spi/impl/ProtocolProcessorTest.java | 4 +- 2 files changed, 47 insertions(+), 12 deletions(-) diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java index a23208dd..cee54bbd 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java @@ -14,43 +14,77 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** */ public class DummyStorageService implements IStorageService { private Map> m_persistentSubscriptions = new HashMap>(); + private Map m_retainedStore = new HashMap(); + private Map> m_persistentMessageStore = new HashMap>(); + private Map m_inflightStore = new HashMap(); + private Map m_qos2Store = new HashMap(); + + private static final Logger LOG = LoggerFactory.getLogger(DummyStorageService.class); public void initStore() { //To change body of implemented methods use File | Settings | File Templates. } public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { - //To change body of implemented methods use File | Settings | File Templates. + if (message.length == 0) { + //clean the message from topic + m_retainedStore.remove(topic); + } else { + //store the message to the topic + m_retainedStore.put(topic, new HawtDBStorageService.StoredMessage(message, qos, topic)); + } } public Collection searchMatching(IMatchingCondition condition) { - return null; //To change body of implemented methods use File | Settings | File Templates. + LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size()); + + List results = new ArrayList(); + + for (Map.Entry entry : m_retainedStore.entrySet()) { + HawtDBStorageService.StoredMessage storedMsg = entry.getValue(); + if (condition.match(entry.getKey())) { + results.add(storedMsg); + } + } + + return results; } public void storePublishForFuture(PublishEvent evt) { - //To change body of implemented methods use File | Settings | File Templates. + LOG.debug("storePublishForFuture store evt " + 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 null; //To change body of implemented methods use File | Settings | File Templates. + return m_persistentMessageStore.get(clientID); } public void cleanPersistedPublishes(String clientID) { - //To change body of implemented methods use File | Settings | File Templates. + m_persistentMessageStore.remove(clientID); } public void cleanInFlight(String msgID) { - //To change body of implemented methods use File | Settings | File Templates. + m_inflightStore.remove(msgID); } public void addInFlight(PublishEvent evt, String publishKey) { - //To change body of implemented methods use File | Settings | File Templates. + m_inflightStore.put(publishKey, evt); } public void addNewSubscription(Subscription newSubscription, String clientID) { @@ -82,14 +116,15 @@ public class DummyStorageService implements IStorageService { } public void persistQoS2Message(String publishKey, PublishEvent evt) { - //To change body of implemented methods use File | Settings | File Templates. + LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); + m_qos2Store.put(publishKey, evt); } public void removeQoS2Message(String publishKey) { - //To change body of implemented methods use File | Settings | File Templates. + m_qos2Store.remove(publishKey); } public PublishEvent retrieveQoS2Message(String publishKey) { - return null; //To change body of implemented methods use File | Settings | File Templates. + return m_qos2Store.get(publishKey); } } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index e4a35d71..88e40325 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -74,8 +74,8 @@ public class ProtocolProcessorTest { //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); - m_storageService = new HawtDBStorageService(); - m_storageService.initStore(); + m_storageService = new DummyStorageService(); + //m_storageService.initStore(); subscriptions = new SubscriptionsStore(); subscriptions.init(m_storageService);