From 213851335a4f22fc30796911d5bebcd1d26fb7c7 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 28 Nov 2018 23:05:24 +0100 Subject: [PATCH] Added implementation of retained store in H2 --- ChangeLog.txt | 2 + .../moquette/broker/IRetainedRepository.java | 4 +- .../broker/MemoryRetainedRepository.java | 26 +++++++-- .../java/io/moquette/broker/PostOffice.java | 19 ++++--- .../io/moquette/broker/RetainedMessage.java | 24 ++++++++ .../main/java/io/moquette/broker/Server.java | 5 +- .../io/moquette/persistence/H2Builder.java | 5 ++ .../persistence/H2RetainedRepository.java | 55 +++++++++++++++++++ .../broker/PostOfficePublishTest.java | 10 ++-- 9 files changed, 131 insertions(+), 19 deletions(-) create mode 100644 broker/src/main/java/io/moquette/broker/RetainedMessage.java create mode 100644 broker/src/main/java/io/moquette/persistence/H2RetainedRepository.java diff --git a/ChangeLog.txt b/ChangeLog.txt index 3a84a44f..f18fbf62 100644 --- a/ChangeLog.txt +++ b/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 diff --git a/broker/src/main/java/io/moquette/broker/IRetainedRepository.java b/broker/src/main/java/io/moquette/broker/IRetainedRepository.java index 11af4d54..5f172832 100644 --- a/broker/src/main/java/io/moquette/broker/IRetainedRepository.java +++ b/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 retainedOnTopic(String topic); } diff --git a/broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java b/broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java index 206a3784..e991e0c2 100644 --- a/broker/src/main/java/io/moquette/broker/MemoryRetainedRepository.java +++ b/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 storage = new ConcurrentHashMap<>(); + private final ConcurrentMap 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) { - return storage.get(topic); + public List retainedOnTopic(String topic) { + final Topic searchTopic = new Topic(topic); + final List matchingMessages = new ArrayList<>(); + for (Map.Entry entry : storage.entrySet()) { + final Topic scanTopic = entry.getKey(); + if (searchTopic.match(scanTopic)) { + matchingMessages.add(entry.getValue()); + } + } + return matchingMessages; } } diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java index 6119b0fc..08a61096 100644 --- a/broker/src/main/java/io/moquette/broker/PostOffice.java +++ b/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 retainedMsgs = retainedRepository.retainedOnTopic(topicFilter); - if (retainedMsg == null) { + if (retainedMsgs.isEmpty()) { // not found continue; } + for (RetainedMessage retainedMsg : retainedMsgs) { + final MqttQoS retainedQos = retainedMsg.qosLevel(); + MqttQoS qos = lowerQosToTheSubscriptionDesired(subscription, retainedQos); - final MqttQoS retainedQos = retainedMsg.fixedHeader().qosLevel(); - MqttQoS qos = lowerQosToTheSubscriptionDesired(subscription, retainedQos); - - final ByteBuf origPayload = retainedMsg.payload(); - ByteBuf payload = origPayload.retainedDuplicate(); - targetSession.sendRetainedPublishOnSessionAtQos(subscription.getTopicFilter(), qos, payload); +// final ByteBuf origPayload = retainedMsg.getPayload(); + final ByteBuf payloadBuf = Unpooled.wrappedBuffer(retainedMsg.getPayload()); +// ByteBuf payload = origPayload.retainedDuplicate(); + targetSession.sendRetainedPublishOnSessionAtQos(subscription.getTopicFilter(), qos, payloadBuf); + } } } diff --git a/broker/src/main/java/io/moquette/broker/RetainedMessage.java b/broker/src/main/java/io/moquette/broker/RetainedMessage.java new file mode 100644 index 00000000..6190a691 --- /dev/null +++ b/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; + } +} diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index 4ae6c828..610e5e3d 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/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, diff --git a/broker/src/main/java/io/moquette/persistence/H2Builder.java b/broker/src/main/java/io/moquette/persistence/H2Builder.java index c82a0aeb..fd72e757 100644 --- a/broker/src/main/java/io/moquette/persistence/H2Builder.java +++ b/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); + } } diff --git a/broker/src/main/java/io/moquette/persistence/H2RetainedRepository.java b/broker/src/main/java/io/moquette/persistence/H2RetainedRepository.java new file mode 100644 index 00000000..0fd92adf --- /dev/null +++ b/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 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 retainedOnTopic(String topic) { + final Topic searchTopic = new Topic(topic); + final List matchingMessages = new ArrayList<>(); + for (Map.Entry entry : queueMap.entrySet()) { + final Topic scanTopic = entry.getKey(); + if (searchTopic.match(scanTopic)) { + matchingMessages.add(entry.getValue()); + } + } + + return matchingMessages; + } +} diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java index aa740a40..bd898df9 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java +++ b/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) { - MqttPublishMessage msg = retainedRepository.retainedOnTopic(expectedTopicName); - assertEquals(expectedTopicName, msg.variableHeader().topicName()); + private void assertMessageIsRetained(String expectedTopicName, ByteBuf expectedPayload) { + List msgs = retainedRepository.retainedOnTopic(expectedTopicName); + assertEquals(1, msgs.size()); + RetainedMessage msg = msgs.get(0); + assertEquals(ByteBufUtil.hexDump(expectedPayload), ByteBufUtil.hexDump(msg.getPayload())); } }