Added implementation of retained store in H2

ChangeLog.txt

@@ -1,6 +1,8 @@
 Version 0.12:
    Fixed issue #415
    Dropped Hazelcast because at the moment Moquette is not designed to be clusterizable
+   Rewritten completly the broker to introduce the concept of MqttConnection
+   Dropped suppoert for MapDB and keeped only H2 MvStore for persistence
 
 Version 0.10:
     Netty epoll system added

broker/src/main/java/io/moquette/broker/IRetainedRepository.java

@@ -18,6 +18,8 @@ package io.moquette.broker;
 import io.moquette.broker.subscriptions.Topic;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
 
+import java.util.List;
+
 public interface IRetainedRepository {
 
     void cleanRetained(Topic topic);
@@ -26,5 +28,5 @@ public interface IRetainedRepository {
 
     boolean isEmpty();
 
-    MqttPublishMessage retainedOnTopic(String topic);
+    List<RetainedMessage> retainedOnTopic(String topic);
 }

broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java

@@ -16,8 +16,12 @@
 package io.moquette.broker;
 
 import io.moquette.broker.subscriptions.Topic;
+import io.netty.buffer.ByteBuf;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -26,16 +30,20 @@ import java.util.concurrent.ConcurrentMap;
 * */
 final class MemoryRetainedRepository implements IRetainedRepository {
 
-    private final ConcurrentMap<String, MqttPublishMessage> storage = new ConcurrentHashMap<>();
+    private final ConcurrentMap<Topic, RetainedMessage> storage = new ConcurrentHashMap<>();
 
     @Override
     public void cleanRetained(Topic topic) {
-        storage.remove(topic.toString());
+        storage.remove(topic);
     }
 
     @Override
     public void retain(Topic topic, MqttPublishMessage msg) {
-        storage.put(topic.toString(), msg);
+        final ByteBuf payload = msg.content();
+        byte[] rawPayload = new byte[payload.readableBytes()];
+        payload.getBytes(0, rawPayload);
+        final RetainedMessage toStore = new RetainedMessage(msg.fixedHeader().qosLevel(), rawPayload);
+        storage.put(topic, toStore);
     }
 
     @Override
@@ -44,7 +52,15 @@ final class MemoryRetainedRepository implements IRetainedRepository {
     }
 
     @Override
-    public MqttPublishMessage retainedOnTopic(String topic) {
+    public List<RetainedMessage> retainedOnTopic(String topic) {
-        return storage.get(topic);
+        final Topic searchTopic = new Topic(topic);
+        final List<RetainedMessage> matchingMessages = new ArrayList<>();
+        for (Map.Entry<Topic, RetainedMessage> entry : storage.entrySet()) {
+            final Topic scanTopic = entry.getKey();
+            if (searchTopic.match(scanTopic)) {
+                matchingMessages.add(entry.getValue());
+            }
+        }
+        return matchingMessages;
     }
 }

broker/src/main/java/io/moquette/broker/PostOffice.java

@@ -21,6 +21,7 @@ import io.moquette.broker.subscriptions.Subscription;
 import io.moquette.broker.subscriptions.Topic;
 import io.moquette.broker.security.IAuthorizatorPolicy;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.mqtt.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -99,19 +100,21 @@ class PostOffice {
         Session targetSession = this.sessionRegistry.retrieve(clientID);
         for (Subscription subscription : newSubscriptions) {
             final String topicFilter = subscription.getTopicFilter().toString();
-            final MqttPublishMessage retainedMsg = retainedRepository.retainedOnTopic(topicFilter);
+            final List<RetainedMessage> retainedMsgs = retainedRepository.retainedOnTopic(topicFilter);
 
-            if (retainedMsg == null) {
+            if (retainedMsgs.isEmpty()) {
                 // not found
                 continue;
             }
-
+            for (RetainedMessage retainedMsg : retainedMsgs) {
-            final MqttQoS retainedQos = retainedMsg.fixedHeader().qosLevel();
+                final MqttQoS retainedQos = retainedMsg.qosLevel();
                 MqttQoS qos = lowerQosToTheSubscriptionDesired(subscription, retainedQos);
 
-            final ByteBuf origPayload = retainedMsg.payload();
+//                final ByteBuf origPayload = retainedMsg.getPayload();
-            ByteBuf payload = origPayload.retainedDuplicate();
+                final ByteBuf payloadBuf = Unpooled.wrappedBuffer(retainedMsg.getPayload());
-            targetSession.sendRetainedPublishOnSessionAtQos(subscription.getTopicFilter(), qos, payload);
+//                ByteBuf payload = origPayload.retainedDuplicate();
+                targetSession.sendRetainedPublishOnSessionAtQos(subscription.getTopicFilter(), qos, payloadBuf);
+            }
         }
     }
 

broker/src/main/java/io/moquette/broker/RetainedMessage.java

@@ -0,0 +1,24 @@
+package io.moquette.broker;
+
+import io.netty.handler.codec.mqtt.MqttQoS;
+
+import java.io.Serializable;
+
+public class RetainedMessage implements Serializable{
+
+    private final MqttQoS qos;
+    private final byte[] payload;
+
+    public RetainedMessage(MqttQoS qos, byte[] payload) {
+        this.qos = qos;
+        this.payload = payload;
+    }
+
+    public MqttQoS qosLevel() {
+        return qos;
+    }
+
+    public byte[] getPayload() {
+        return payload;
+    }
+}

broker/src/main/java/io/moquette/broker/Server.java

@@ -162,21 +162,24 @@ public class Server {
 
         final ISubscriptionsRepository subscriptionsRepository;
         final IQueueRepository queueRepository;
+        final IRetainedRepository retainedRepository;
         if (persistencePath != null && !persistencePath.isEmpty()) {
             LOG.trace("Configuring H2 subscriptions store to {}", persistencePath);
             h2Builder = new H2Builder(config, scheduler).initStore();
             subscriptionsRepository = h2Builder.subscriptionsRepository();
             queueRepository = h2Builder.queueRepository();
+            retainedRepository = h2Builder.retainedRepository();
         } else {
             LOG.trace("Configuring in-memory subscriptions store");
             subscriptionsRepository = new MemorySubscriptionsRepository();
             queueRepository = new MemoryQueueRepository();
+            retainedRepository = new MemoryRetainedRepository();
         }
 
         ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
         subscriptions.init(subscriptionsRepository);
         SessionRegistry sessions = new SessionRegistry(subscriptions, queueRepository);
-        dispatcher = new PostOffice(subscriptions, authorizatorPolicy, new MemoryRetainedRepository(), sessions,
+        dispatcher = new PostOffice(subscriptions, authorizatorPolicy, retainedRepository, sessions,
                                     interceptor);
         final BrokerConfiguration brokerConfig = new BrokerConfiguration(config);
         MQTTConnectionFactory connectionFactory = new MQTTConnectionFactory(brokerConfig, authenticator, sessions,

broker/src/main/java/io/moquette/persistence/H2Builder.java

@@ -2,6 +2,7 @@ package io.moquette.persistence;
 
 import io.moquette.BrokerConstants;
 import io.moquette.broker.IQueueRepository;
+import io.moquette.broker.IRetainedRepository;
 import io.moquette.broker.ISubscriptionsRepository;
 import io.moquette.broker.config.IConfig;
 import org.h2.mvstore.MVStore;
@@ -57,4 +58,8 @@ public class H2Builder {
     public IQueueRepository queueRepository() {
         return new H2QueueRepository(mvStore);
     }
+
+    public IRetainedRepository retainedRepository() {
+        return new H2RetainedRepository(mvStore);
+    }
 }

broker/src/main/java/io/moquette/persistence/H2RetainedRepository.java

@@ -0,0 +1,55 @@
+package io.moquette.persistence;
+
+import io.moquette.broker.IRetainedRepository;
+import io.moquette.broker.RetainedMessage;
+import io.moquette.broker.subscriptions.Topic;
+import io.netty.buffer.ByteBuf;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import org.h2.mvstore.MVMap;
+import org.h2.mvstore.MVStore;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class H2RetainedRepository implements IRetainedRepository {
+
+    private final MVMap<Topic, RetainedMessage> queueMap;
+
+    public H2RetainedRepository(MVStore mvStore) {
+        this.queueMap = mvStore.openMap("retained_store");
+    }
+
+    @Override
+    public void cleanRetained(Topic topic) {
+        queueMap.remove(topic);
+    }
+
+    @Override
+    public void retain(Topic topic, MqttPublishMessage msg) {
+        final ByteBuf payload = msg.content();
+        byte[] rawPayload = new byte[payload.readableBytes()];
+        payload.getBytes(0, rawPayload);
+        final RetainedMessage toStore = new RetainedMessage(msg.fixedHeader().qosLevel(), rawPayload);
+        queueMap.put(topic, toStore);
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return queueMap.isEmpty();
+    }
+
+    @Override
+    public List<RetainedMessage> retainedOnTopic(String topic) {
+        final Topic searchTopic = new Topic(topic);
+        final List<RetainedMessage> matchingMessages = new ArrayList<>();
+        for (Map.Entry<Topic, RetainedMessage> entry : queueMap.entrySet()) {
+            final Topic scanTopic = entry.getKey();
+            if (searchTopic.match(scanTopic)) {
+                matchingMessages.add(entry.getValue());
+            }
+        }
+
+        return matchingMessages;
+    }
+}

broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java

@@ -390,7 +390,7 @@ public class PostOfficePublishTest {
         sut.receivedPublishQos1(connection, new Topic(NEWS_TOPIC), TEST_USER, anyPayload, 1, true,
                                 publishMsg);
 
-        assertMessageIsRetained(NEWS_TOPIC);
+        assertMessageIsRetained(NEWS_TOPIC, anyPayload);
 
         // publish a QoS0 retained message
         // Exercise
@@ -406,8 +406,10 @@ public class PostOfficePublishTest {
         assertTrue("Retained message for topic /news must be cleared", retainedRepository.isEmpty());
     }
 
-    private void assertMessageIsRetained(String expectedTopicName) {
+    private void assertMessageIsRetained(String expectedTopicName, ByteBuf expectedPayload) {
-        MqttPublishMessage msg = retainedRepository.retainedOnTopic(expectedTopicName);
+        List<RetainedMessage> msgs = retainedRepository.retainedOnTopic(expectedTopicName);
-        assertEquals(expectedTopicName, msg.variableHeader().topicName());
+        assertEquals(1, msgs.size());
+        RetainedMessage msg = msgs.get(0);
+        assertEquals(ByteBufUtil.hexDump(expectedPayload), ByteBufUtil.hexDump(msg.getPayload()));
     }
 }