Completed the memory implementation of DummyStorageService

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<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
+    private Map<String, HawtDBStorageService.StoredMessage> m_retainedStore = new HashMap<String, HawtDBStorageService.StoredMessage>();
+    private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<String, List<PublishEvent>>();
+    private Map<String, PublishEvent> m_inflightStore = new HashMap<String, PublishEvent>();
+    private Map<String, PublishEvent> m_qos2Store = new HashMap<String, PublishEvent>();
+    
+    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<HawtDBStorageService.StoredMessage> 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<HawtDBStorageService.StoredMessage> results = new ArrayList<HawtDBStorageService.StoredMessage>();
+
+        for (Map.Entry<String, HawtDBStorageService.StoredMessage> 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<PublishEvent> storedEvents;
+        String clientID = evt.getClientID();
+        if (!m_persistentMessageStore.containsKey(clientID)) {
+            storedEvents = new ArrayList<PublishEvent>();
+        } else {
+            storedEvents = m_persistentMessageStore.get(clientID);
+        }
+        storedEvents.add(evt);
+        m_persistentMessageStore.put(clientID, storedEvents);
     }
 
     public List<PublishEvent> 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);
     }
 }

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);