From d9ff81131c69ba46f993c63efb0c5e4615e6e170 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 20 Apr 2017 23:14:23 +0200 Subject: [PATCH] Merged topic subscription search and publish into single MessagePublisher method First step on implementing H2 storage Minor, fixed not imported class Cleaned MessageStore interface to avoid duplication of methods used in couple. Switched from guid + message to move messages Separated ISubscriptionsStore from ISessionStore Implemented H2 queue --- broker/pom.xml | 14 + .../persistence/MemoryMessagesStore.java | 45 +-- .../persistence/MemorySessionStore.java | 165 ++++---- .../persistence/MemoryStorageService.java | 6 +- .../java/io/moquette/spi/ClientSession.java | 70 ++-- .../java/io/moquette/spi/IMessagesStore.java | 25 +- .../java/io/moquette/spi/ISessionsStore.java | 107 +---- .../src/main/java/io/moquette/spi/IStore.java | 18 +- .../io/moquette/spi/ISubscriptionsStore.java | 90 +++++ .../moquette/spi/impl/MessagesPublisher.java | 10 +- .../moquette/spi/impl/ProtocolProcessor.java | 59 ++- .../impl/ProtocolProcessorBootstrapper.java | 27 +- .../moquette/spi/impl/Qos1PublishHandler.java | 11 +- .../moquette/spi/impl/Qos2PublishHandler.java | 35 +- .../spi/impl/subscriptions/Subscription.java | 2 +- ...Store.java => SubscriptionsDirectory.java} | 23 +- .../spi/impl/subscriptions/TreeNode.java | 2 +- .../moquette/persistence/MessageStoreTCK.java | 92 +++++ .../io/moquette/spi/ClientSessionTest.java | 6 +- .../spi/impl/ProtocolProcessorTest.java | 18 +- .../impl/ProtocolProcessor_CONNECT_Test.java | 6 +- ...t.java => SubscriptionsDirectoryTest.java} | 80 ++-- .../spi/impl/subscriptions/TreeNodeTest.java | 10 +- broker/src/test/resources/log4j.properties | 2 +- .../src/main/resources/log4j.properties | 4 +- h2_storage/pom.xml | 129 ++++++ .../persistence/h2/H2MessagesStore.java | 93 +++++ .../persistence/h2/H2PersistentQueue.java | 83 ++++ .../persistence/h2/H2PersistentStore.java | 94 +++++ .../persistence/h2/H2SessionsStore.java | 370 ++++++++++++++++++ .../persistence/h2/H2MessagesStoreTest.java | 69 ++++ .../persistence/h2/H2PersistentQueueTest.java | 96 +++++ mapdb_storage/pom.xml | 49 +-- .../persistence/mapdb/MapDBMessagesStore.java | 65 +-- .../mapdb/MapDBPersistentStore.java | 43 +- .../persistence/mapdb/MapDBSessionsStore.java | 199 +++++----- .../mapdb/MapDBPersistentStoreTest.java | 98 +---- .../impl/subscriptions/PerformanceTest.java | 4 +- pom.xml | 1 + 39 files changed, 1585 insertions(+), 735 deletions(-) create mode 100644 broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java rename broker/src/main/java/io/moquette/spi/impl/subscriptions/{SubscriptionsStore.java => SubscriptionsDirectory.java} (91%) create mode 100644 broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java rename broker/src/test/java/io/moquette/spi/impl/subscriptions/{SubscriptionsStoreTest.java => SubscriptionsDirectoryTest.java} (83%) create mode 100644 h2_storage/pom.xml create mode 100644 h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java create mode 100644 h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java create mode 100644 h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java create mode 100644 h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java create mode 100644 h2_storage/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java create mode 100644 h2_storage/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java diff --git a/broker/pom.xml b/broker/pom.xml index 3a67af93..322f8283 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -173,6 +173,20 @@ + + + org.apache.maven.plugins + maven-jar-plugin + 3.0.2 + + + + test-jar + + + + + maven-jar-plugin diff --git a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java index cfde2918..1d3ef0fb 100644 --- a/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryMessagesStore.java @@ -16,20 +16,19 @@ package io.moquette.persistence; -import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMatchingCondition; -import io.moquette.spi.MessageGUID; +import io.moquette.spi.IMessagesStore; import io.moquette.spi.impl.subscriptions.Topic; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.util.*; public class MemoryMessagesStore implements IMessagesStore { private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class); - private Map m_retainedStore = new HashMap<>(); - private Map m_persistentMessageStore = new HashMap<>(); + private Map m_retainedStore = new HashMap<>(); MemoryMessagesStore() { } @@ -39,8 +38,12 @@ public class MemoryMessagesStore implements IMessagesStore { } @Override - public void storeRetained(Topic topic, MessageGUID guid) { - m_retainedStore.put(topic, guid); + public void storeRetained(Topic topic, StoredMessage storedMessage) { + LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID()); + if (storedMessage.getClientID() == null) { + throw new IllegalArgumentException( "Message to be persisted must have a not null client ID"); + } + m_retainedStore.put(topic, storedMessage); } @Override @@ -49,9 +52,8 @@ public class MemoryMessagesStore implements IMessagesStore { List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - final MessageGUID guid = entry.getValue(); - StoredMessage storedMsg = m_persistentMessageStore.get(guid); + for (Map.Entry entry : m_retainedStore.entrySet()) { + StoredMessage storedMsg = entry.getValue(); if (condition.match(entry.getKey())) { results.add(storedMsg); } @@ -60,31 +62,6 @@ public class MemoryMessagesStore implements IMessagesStore { return results; } - @Override - public MessageGUID storePublishForFuture(StoredMessage storedMessage) { - LOG.debug("storePublishForFuture store evt {}", storedMessage); - MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); - storedMessage.setGuid(guid); - m_persistentMessageStore.put(guid, storedMessage); - return guid; - } - - @Override - public void dropInFlightMessagesInSession(Collection pendingAckMessages) { - //remove all guids from retained - Collection messagesToRemove = new HashSet<>(pendingAckMessages); - messagesToRemove.removeAll(m_retainedStore.values()); - - for (MessageGUID guid : messagesToRemove) { - m_persistentMessageStore.remove(guid); - } - } - - @Override - public StoredMessage getMessageByGuid(MessageGUID guid) { - return m_persistentMessageStore.get(guid); - } - @Override public void cleanRetained(Topic topic) { m_retainedStore.remove(topic); diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index bba39dc2..3a385bc7 100644 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -18,22 +18,20 @@ package io.moquette.persistence; import io.moquette.server.Constants; import io.moquette.spi.ClientSession; -import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.MessageGUID; +import io.moquette.spi.ISubscriptionsStore; import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.util.*; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; -import static io.moquette.spi.impl.Utils.defaultGet; - -public class MemorySessionStore implements ISessionsStore { +public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore { private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); @@ -41,21 +39,18 @@ public class MemorySessionStore implements ISessionsStore { private Map m_persistentSessions = new HashMap<>(); - // maps clientID->[MessageId -> guid] - private Map> m_inflightStore = new HashMap<>(); // maps clientID->BlockingQueue private Map> queues = new HashMap<>(); - // maps clientID->[MessageId -> guid] - private Map> m_secondPhaseStore = new HashMap<>(); + // maps clientID->[MessageId -> msg] + private Map> m_secondPhaseStore = new HashMap<>(); - private Map> outboundFlightMessageToGuid = new HashMap<>(); + // maps clientID->[MessageId -> msg] + private Map> outboundFlightMessages = new HashMap<>(); - private Map> inboundFlightMessageToGuid = new HashMap<>(); + // maps clientID->[MessageId -> msg] + private Map> inboundFlightMessages = new HashMap<>(); - private final IMessagesStore m_messagesStore; - - public MemorySessionStore(IMessagesStore messagesStore) { - this.m_messagesStore = messagesStore; + public MemorySessionStore() { } @Override @@ -72,6 +67,11 @@ public class MemorySessionStore implements ISessionsStore { public void initStore() { } + @Override + public ISubscriptionsStore subscriptionStore() { + return this; + } + @Override public void addNewSubscription(Subscription newSubscription) { final String clientID = newSubscription.getClientId(); @@ -102,7 +102,7 @@ public class MemorySessionStore implements ISessionsStore { LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); m_persistentSubscriptions.put(clientID, new HashMap()); m_persistentSessions.put(clientID, new PersistentSession(cleanSession)); - return new ClientSession(clientID, m_messagesStore, this, cleanSession); + return new ClientSession(clientID, this, this, cleanSession); } @Override @@ -112,14 +112,14 @@ public class MemorySessionStore implements ISessionsStore { } PersistentSession storedSession = m_persistentSessions.get(clientID); - return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); + return new ClientSession(clientID, this, this, storedSession.cleanSession); } @Override public Collection getAllSessions() { Collection result = new ArrayList<>(); for (Map.Entry entry : m_persistentSessions.entrySet()) { - result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); + result.add(new ClientSession(entry.getKey(), this, this, entry.getValue().cleanSession)); } return result; } @@ -159,28 +159,25 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public void inFlightAck(String clientID, int messageID) { - Map m = this.m_inflightStore.get(clientID); + public StoredMessage inFlightAck(String clientID, int messageID) { + Map m = this.outboundFlightMessages.get(clientID); if (m == null) { LOG.error("Can't find the inFlight record for client <{}>", clientID); - return; + throw new RuntimeException("Can't find the inFlight record for client <" + clientID + ">"); } - m.remove(messageID); + StoredMessage msg = m.remove(messageID); + this.outboundFlightMessages.put(clientID, m); + return msg; } @Override - public void inFlight(String clientID, int messageID, MessageGUID guid) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - m = new HashMap<>(); + public void inFlight(String clientID, int messageID, StoredMessage msg) { + Map messages = outboundFlightMessages.get(clientID); + if (messages == null) { + messages = new HashMap<>(); } - m.put(messageID, guid); - this.m_inflightStore.put(clientID, m); - - final HashMap emptyGuids = new HashMap<>(); - Map guids = defaultGet(outboundFlightMessageToGuid, clientID, emptyGuids); - guids.put(messageID, guid); - outboundFlightMessageToGuid.put(clientID, guids); + messages.put(messageID, msg); + outboundFlightMessages.put(clientID, messages); } /** @@ -188,7 +185,7 @@ public class MemorySessionStore implements ISessionsStore { */ @Override public int nextPacketID(String clientID) { - Map m = this.m_inflightStore.get(clientID); + Map m = this.outboundFlightMessages.get(clientID); if (m == null) { m = new HashMap<>(); int nextPacketId = 1; @@ -215,61 +212,73 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { - LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); - Map m = this.m_inflightStore.get(clientID); + public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) { + LOG.info("Moving msg inflight second phase store, clientID <{}> messageID {}", clientID, messageID); + Map m = this.m_secondPhaseStore.get(clientID); if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); - return; + m = new HashMap<>(); } - MessageGUID guid = m.remove(messageID); - - LOG.info("Moving to second phase store"); - final HashMap emptyGuids = new HashMap<>(); - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); - messageIDs.put(messageID, guid); - m_secondPhaseStore.put(clientID, messageIDs); + m.put(messageID, msg); + this.outboundFlightMessages.put(clientID, m); } @Override - public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { - final HashMap emptyGuids = new HashMap<>(); - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); - MessageGUID guid = messageIDs.remove(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - return guid; - } + public StoredMessage secondPhaseAcknowledged(String clientID, int messageID) { + LOG.info("Acknowledged message in second phase, clientID <{}> messageID {}", clientID, messageID); + final Map m = this.m_secondPhaseStore.get(clientID); + if (m == null) { + String error = String.format("Can't find the inFlight record for client <%s> during the second phase " + + "acking of QoS2 pub", clientID); + LOG.error(error); + throw new RuntimeException(error); + } - @Override - public StoredMessage getInflightMessage(String clientID, int messageID) { - Map inflightMessages = m_inflightStore.get(clientID); - MessageGUID guid = inflightMessages.get(messageID); - return m_messagesStore.getMessageByGuid(guid); + StoredMessage msg = m.remove(messageID); + m_secondPhaseStore.put(clientID, m); + return msg; } @Override public int getInflightMessagesNo(String clientID) { - Map inflightMessages = m_inflightStore.get(clientID); - if (inflightMessages == null) - return 0; - else - return inflightMessages.size(); + int totalInflight = 0; + Map inflightPerClient = this.inboundFlightMessages.get(clientID); + if (inflightPerClient != null) { + totalInflight += inflightPerClient.size(); + } + + Map secondPhaseInFlight = this.m_secondPhaseStore.get(clientID); + if (secondPhaseInFlight != null) { + totalInflight += secondPhaseInFlight.size(); + } + + Map outboundPerClient = outboundFlightMessages.get(clientID); + if (outboundPerClient != null) { + totalInflight += outboundPerClient.size(); + } + + return totalInflight; } @Override public StoredMessage inboundInflight(String clientID, int messageID) { - final HashMap emptyGuids = new HashMap<>(); - Map guids = Utils.defaultGet(inboundFlightMessageToGuid, clientID, emptyGuids); - final MessageGUID guid = guids.get(messageID); - return m_messagesStore.getMessageByGuid(guid); + Map inflightPerClient = this.inboundFlightMessages.get(clientID); + if (inflightPerClient == null) { + String error = String.format("Can't find inbound inflight zone for client <%s>", clientID); + LOG.error(error); + throw new RuntimeException(error); + } + return inflightPerClient.get(messageID); } @Override - public void markAsInboundInflight(String clientID, int messageID, MessageGUID guid) { - final HashMap emptyGuids = new HashMap<>(); - Map guids = Utils.defaultGet(inboundFlightMessageToGuid, clientID, emptyGuids); - guids.put(messageID, guid); - inboundFlightMessageToGuid.put(clientID, guids); + public void markAsInboundInflight(String clientID, int messageID, StoredMessage msg) { + Map inflightPerClient = this.inboundFlightMessages.get(clientID); + if (inflightPerClient == null) { + inflightPerClient = new HashMap<>(); + } + + inflightPerClient.put(messageID, msg); + inboundFlightMessages.put(clientID, inflightPerClient); } @Override @@ -279,7 +288,7 @@ public class MemorySessionStore implements ISessionsStore { @Override public int getSecondPhaseAckPendingMessages(String clientID) { - Map pendingAcks = m_secondPhaseStore.get(clientID); + Map pendingAcks = m_secondPhaseStore.get(clientID); if (pendingAcks == null) return 0; else @@ -287,12 +296,8 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public Collection pendingAck(String clientID) { - Map messageGUIDMap = outboundFlightMessageToGuid.get(clientID); - if (messageGUIDMap == null || messageGUIDMap.isEmpty()) { - return Collections.emptyList(); - } - - return new ArrayList<>(messageGUIDMap.values()); + public void dropInFlightMessagesInSession(String clientID) { + outboundFlightMessages.remove(clientID); + inboundFlightMessages.remove(clientID); } } diff --git a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java index 225477cf..05b37a52 100644 --- a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java @@ -27,7 +27,7 @@ public class MemoryStorageService implements IStore { public MemoryStorageService() { m_messagesStore = new MemoryMessagesStore(); - m_sessionsStore = new MemorySessionStore(m_messagesStore); + m_sessionsStore = new MemorySessionStore(); m_messagesStore.initStore(); m_sessionsStore.initStore(); } @@ -42,6 +42,10 @@ public class MemoryStorageService implements IStore { return m_sessionsStore; } + @Override + public void initStore() { + } + @Override public void close() { } diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index d31d5c2d..42a19e65 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -16,11 +16,12 @@ package io.moquette.spi; -import io.moquette.spi.ISessionsStore.ClientTopicCouple; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.util.HashSet; import java.util.Set; import java.util.concurrent.BlockingQueue; @@ -44,22 +45,22 @@ public class ClientSession { class OutboundFlightZone { /** - * Save the binding messageID, clientID - guid - + * Save the binding messageID, clientID - message + * * @param messageID the packet ID used in transmission - * @param guid the guid of the message being in flight. + * @param msg the message to put in flight zone */ - void waitingAck(int messageID, MessageGUID guid) { + void waitingAck(int messageID, IMessagesStore.StoredMessage msg) { if (LOG.isTraceEnabled()) { - LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid); + LOG.trace("Adding to inflight {}, guid <{}>", messageID, msg.getGuid()); } - m_sessionsStore.inFlight(ClientSession.this.clientID, messageID, guid); + m_sessionsStore.inFlight(ClientSession.this.clientID, messageID, msg); } - void acknowledged(int messageID) { + IMessagesStore.StoredMessage acknowledged(int messageID) { if (LOG.isTraceEnabled()) LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", ClientSession.this.clientID, messageID); - m_sessionsStore.inFlightAck(ClientSession.this.clientID, messageID); + return m_sessionsStore.inFlightAck(ClientSession.this.clientID, messageID); } } @@ -69,8 +70,8 @@ public class ClientSession { return m_sessionsStore.inboundInflight(clientID, messageID); } - public void waitingRel(int messageID, MessageGUID guid) { - m_sessionsStore.markAsInboundInflight(clientID, messageID, guid); + public void waitingRel(int messageID, IMessagesStore.StoredMessage msg) { + m_sessionsStore.markAsInboundInflight(clientID, messageID, msg); } } @@ -78,12 +79,11 @@ public class ClientSession { public final String clientID; - private final IMessagesStore messagesStore; - private final ISessionsStore m_sessionsStore; private Set subscriptions = new HashSet<>(); + private final ISubscriptionsStore subscriptionsStore; private volatile boolean cleanSession; // private BlockingQueue m_queueToPublish = new @@ -91,11 +91,11 @@ public class ClientSession { private final OutboundFlightZone outboundFlightZone; private final InboundFlightZone inboundFlightZone; - public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore, - boolean cleanSession) { + public ClientSession(String clientID, ISessionsStore sessionsStore, ISubscriptionsStore subscriptionsStore, + boolean cleanSession) { this.clientID = clientID; - this.messagesStore = messagesStore; this.m_sessionsStore = sessionsStore; + this.subscriptionsStore = subscriptionsStore; this.cleanSession = cleanSession; this.outboundFlightZone = new OutboundFlightZone(); this.inboundFlightZone = new InboundFlightZone(); @@ -129,7 +129,7 @@ public class ClientSession { return false; } ClientTopicCouple matchingCouple = new ClientTopicCouple(this.clientID, newSubscription.getTopicFilter()); - Subscription existingSub = m_sessionsStore.getSubscription(matchingCouple); + Subscription existingSub = subscriptionsStore.getSubscription(matchingCouple); // update the selected subscriptions if not present or if has a greater qos if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) { if (existingSub != null) { @@ -139,14 +139,14 @@ public class ClientSession { subscriptions.remove(newSubscription); } subscriptions.add(newSubscription); - m_sessionsStore.addNewSubscription(newSubscription); + subscriptionsStore.addNewSubscription(newSubscription); } return true; } public void unsubscribeFrom(Topic topicFilter) { LOG.info("Removing subscription. ClientID={}, topics={}", clientID, topicFilter); - m_sessionsStore.removeSubscription(topicFilter, clientID); + subscriptionsStore.removeSubscription(topicFilter, clientID); Set subscriptionsToRemove = new HashSet<>(); for (Subscription sub : this.subscriptions) { if (sub.getTopicFilter().equals(topicFilter)) { @@ -166,11 +166,11 @@ public class ClientSession { public void cleanSession() { LOG.info("Wiping existing subscriptions. ClientId={}", this.clientID); - m_sessionsStore.wipeSubscriptions(this.clientID); + subscriptionsStore.wipeSubscriptions(this.clientID); // remove also the messages stored of type QoS1/2 LOG.info("Removing stored messages with QoS 1 and 2. ClientId={}", this.clientID); - messagesStore.dropInFlightMessagesInSession(m_sessionsStore.pendingAck(this.clientID)); + m_sessionsStore.dropInFlightMessagesInSession(this.clientID); //remove also the enqueued messages this.m_sessionsStore.dropQueue(this.clientID); @@ -189,8 +189,8 @@ public class ClientSession { return this.m_sessionsStore.nextPacketID(this.clientID); } - public void inFlightAcknowledged(int messageID) { - outboundFlightZone.acknowledged(messageID); + public IMessagesStore.StoredMessage inFlightAcknowledged(int messageID) { + return outboundFlightZone.acknowledged(messageID); } /** @@ -200,21 +200,13 @@ public class ClientSession { * */ public int inFlightAckWaiting(IMessagesStore.StoredMessage msg) { LOG.debug("Adding message ot inflight zone. ClientId={}", clientID); - MessageGUID guid = msg.getGuid(); - if (guid == null) { - guid = messagesStore.storePublishForFuture(msg); - msg.setGuid(guid); - LOG.debug("Not yet assigned guid to message, updating guid={}", guid); - } - int messageId = ClientSession.this.nextPacketId(); - outboundFlightZone.waitingAck(messageId, guid); + outboundFlightZone.waitingAck(messageId, msg); return messageId; } public IMessagesStore.StoredMessage secondPhaseAcknowledged(int messageID) { - MessageGUID guid = m_sessionsStore.secondPhaseAcknowledged(clientID, messageID); - return messagesStore.getMessageByGuid(guid); + return m_sessionsStore.secondPhaseAcknowledged(clientID, messageID); } /** @@ -231,16 +223,12 @@ public class ClientSession { return inboundFlightZone.lookup(messageID); } - public void markAsInboundInflight(int messageID, MessageGUID guid) { - inboundFlightZone.waitingRel(messageID, guid); + public void markAsInboundInflight(int messageID, IMessagesStore.StoredMessage msg) { + inboundFlightZone.waitingRel(messageID, msg); } - public void moveInFlightToSecondPhaseAckWaiting(int messageID) { - m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID); - } - - public IMessagesStore.StoredMessage getInflightMessage(int messageID) { - return m_sessionsStore.getInflightMessage(clientID, messageID); + public void moveInFlightToSecondPhaseAckWaiting(int messageID, IMessagesStore.StoredMessage msg) { + m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID, msg); } public Set getSubscriptions() { diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index beb6a373..b8da62e1 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -93,16 +93,6 @@ public interface IMessagesStore { */ void initStore(); - /** - * Persist the message. If the message is empty then the topic is cleaned, else it's stored. - * - * @param topic - * for the retained. - * @param guid - * of the message to mark as retained. - */ - void storeRetained(Topic topic, MessageGUID guid); - /** * Return a list of retained messages that satisfy the condition. * @@ -112,18 +102,7 @@ public interface IMessagesStore { */ Collection searchMatching(IMatchingCondition condition); - /** - * Persist the message. - * - * @param storedMessage - * the message to store for future usage. - * @return the unique id in the storage (guid). - */ - MessageGUID storePublishForFuture(StoredMessage storedMessage); - - void dropInFlightMessagesInSession(Collection pendingAckMessages); - - StoredMessage getMessageByGuid(MessageGUID guid); - void cleanRetained(Topic topic); + + void storeRetained(Topic topic, StoredMessage storedMessage); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 61c98d61..839249c3 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -17,11 +17,8 @@ package io.moquette.spi; import io.moquette.spi.IMessagesStore.StoredMessage; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.Topic; import java.util.Collection; -import java.util.List; import java.util.concurrent.BlockingQueue; /** @@ -29,90 +26,12 @@ import java.util.concurrent.BlockingQueue; */ public interface ISessionsStore { - class ClientTopicCouple { - - public final Topic topicFilter; - public final String clientID; - - public ClientTopicCouple(String clientID, Topic topicFilter) { - this.clientID = clientID; - this.topicFilter = topicFilter; - } - - @Override - public boolean equals(Object o) { - if (this == o) - return true; - if (o == null || getClass() != o.getClass()) - return false; - - ClientTopicCouple that = (ClientTopicCouple) o; - - if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null) - return false; - return !(clientID != null ? !clientID.equals(that.clientID) : that.clientID != null); - } - - @Override - public int hashCode() { - int result = topicFilter != null ? topicFilter.hashCode() : 0; - result = 31 * result + (clientID != null ? clientID.hashCode() : 0); - return result; - } - - @Override - public String toString() { - return "ClientTopicCouple{" + "topicFilter='" + topicFilter + '\'' + ", clientID='" + clientID + '\'' + '}'; - } - } - void initStore(); + ISubscriptionsStore subscriptionStore(); + void updateCleanStatus(String clientID, boolean cleanSession); - /** - * Add a new subscription to the session - * - * @param newSubscription - * the subscription to add. - */ - void addNewSubscription(Subscription newSubscription); - - /** - * Removed a specific subscription - * - * @param topic - * the topic of the subscription. - * @param clientID - * the session client. - */ - void removeSubscription(Topic topic, String clientID); - - /** - * Remove all the subscriptions of the session - * - * @param sessionID - * the client ID - */ - void wipeSubscriptions(String sessionID); - - /** - * Return all topic filters to recreate the subscription tree. - */ - List listAllSubscriptions(); - - /** - * @param couple - * the subscription descriptor. - * @return the subscription stored by clientID and topicFilter, if any else null; - */ - Subscription getSubscription(ClientTopicCouple couple); - - /* - * @return all subscriptions stored. - */ - List getSubscriptions(); - /** * @param clientID * the session client ID. @@ -132,23 +51,23 @@ public interface ISessionsStore { /** * Returns all the sessions * - * @return + * @return the collection of all stored client sessions. */ Collection getAllSessions(); - void inFlightAck(String clientID, int messageID); + StoredMessage inFlightAck(String clientID, int messageID); /** - * Save the binding messageID, clientID - guid + * Save the message msg with messageID, clientID as in flight * * @param clientID * the client ID * @param messageID * the message ID - * @param guid - * the uuid of the message to mark as inflight. + * @param msg + * the message to put in flight zone */ - void inFlight(String clientID, int messageID, MessageGUID guid); + void inFlight(String clientID, int messageID, StoredMessage msg); /** * Return the next valid packetIdentifier for the given client session. @@ -170,7 +89,7 @@ public interface ISessionsStore { void dropQueue(String clientID); - void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID); + void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg); /** * @param clientID @@ -179,9 +98,7 @@ public interface ISessionsStore { * the message ID that reached the second phase. * @return the guid of message just acked. */ - MessageGUID secondPhaseAcknowledged(String clientID, int messageID); - - StoredMessage getInflightMessage(String clientID, int messageID); + StoredMessage secondPhaseAcknowledged(String clientID, int messageID); /** * Returns the number of inflight messages for the given client ID @@ -196,7 +113,7 @@ public interface ISessionsStore { * */ IMessagesStore.StoredMessage inboundInflight(String clientID, int messageID); - void markAsInboundInflight(String clientID, int messageID, MessageGUID guid); + void markAsInboundInflight(String clientID, int messageID, StoredMessage msg); /** * Returns the size of the session queue for the given client ID @@ -214,5 +131,5 @@ public interface ISessionsStore { */ int getSecondPhaseAckPendingMessages(String clientID); - Collection pendingAck(String clientID); + void dropInFlightMessagesInSession(String clientID); } diff --git a/broker/src/main/java/io/moquette/spi/IStore.java b/broker/src/main/java/io/moquette/spi/IStore.java index 9623d9f0..cfa9764b 100644 --- a/broker/src/main/java/io/moquette/spi/IStore.java +++ b/broker/src/main/java/io/moquette/spi/IStore.java @@ -18,9 +18,21 @@ package io.moquette.spi; public interface IStore { - IMessagesStore messagesStore(); - - ISessionsStore sessionsStore(); + void initStore(); void close(); + + /** + * Factory method to instantiate the messages store. + * + * @return the create instance of IMessagesStore. + * */ + IMessagesStore messagesStore(); + + /** + * Factory method to instantiate the session store. + * + * @return the create instance of ISessionsStore. + * */ + ISessionsStore sessionsStore(); } diff --git a/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java new file mode 100644 index 00000000..d0975de8 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/ISubscriptionsStore.java @@ -0,0 +1,90 @@ +package io.moquette.spi; + +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; + +import java.util.List; + +public interface ISubscriptionsStore { + class ClientTopicCouple { + + public final Topic topicFilter; + public final String clientID; + + public ClientTopicCouple(String clientID, Topic topicFilter) { + this.clientID = clientID; + this.topicFilter = topicFilter; + } + + @Override + public boolean equals(Object o) { + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; + + ClientTopicCouple that = (ClientTopicCouple) o; + + if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null) + return false; + return !(clientID != null ? !clientID.equals(that.clientID) : that.clientID != null); + } + + @Override + public int hashCode() { + int result = topicFilter != null ? topicFilter.hashCode() : 0; + result = 31 * result + (clientID != null ? clientID.hashCode() : 0); + return result; + } + + @Override + public String toString() { + return "ClientTopicCouple{" + "topicFilter='" + topicFilter + '\'' + ", clientID='" + clientID + '\'' + '}'; + } + } + + + /** + * Add a new subscription to the session + * + * @param newSubscription + * the subscription to add. + */ + void addNewSubscription(Subscription newSubscription); + + /** + * Removed a specific subscription + * + * @param topic + * the topic of the subscription. + * @param clientID + * the session client. + */ + void removeSubscription(Topic topic, String clientID); + + /** + * Remove all the subscriptions of the session + * + * @param sessionID + * the client ID + */ + void wipeSubscriptions(String sessionID); + + /** + * Return all topic filters to recreate the subscription tree. + */ + List listAllSubscriptions(); + + /** + * @param couple + * the subscription descriptor. + * @return the subscription stored by clientID and topicFilter, if any else null; + */ + Subscription getSubscription(ClientTopicCouple couple); + + /* + * @return all subscriptions stored. + */ + List getSubscriptions(); + +} diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 34473e15..a10b3d8d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -21,7 +21,7 @@ import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Topic; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.*; @@ -36,10 +36,10 @@ class MessagesPublisher { private final ConnectionDescriptorStore connectionDescriptors; private final ISessionsStore m_sessionsStore; private final PersistentQueueMessageSender messageSender; - private final SubscriptionsStore subscriptions; + private final SubscriptionsDirectory subscriptions; public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, - PersistentQueueMessageSender messageSender, SubscriptionsStore subscriptions) { + PersistentQueueMessageSender messageSender, SubscriptionsDirectory subscriptions) { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; this.messageSender = messageSender; @@ -59,11 +59,11 @@ class MessagesPublisher { void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic, int messageID) { if (LOG.isTraceEnabled()) { - LOG.trace("Sending publish message to subscribers. CId={}, topic={}, messageId={}, payload={}, " + + LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + "subscriptionTree={}", pubMsg.getClientID(), topic, messageID, DebugUtils.payload2Str(pubMsg.getPayload()), subscriptions.dumpTree()); } else { - LOG.info("Sending publish message to subscribers. CId={}, topic={}, messageId={}", pubMsg.getClientID(), topic, + LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", pubMsg.getClientID(), topic, messageID); } publish2Subscribers(pubMsg, topic); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index fb5fc275..5eabfe81 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -24,7 +24,7 @@ import io.moquette.server.netty.NettyUtils; import io.moquette.spi.*; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; @@ -130,7 +130,8 @@ public class ProtocolProcessor { protected ConnectionDescriptorStore connectionDescriptors; protected ConcurrentMap subscriptionInCourse; - private SubscriptionsStore subscriptions; + private SubscriptionsDirectory subscriptions; + private ISubscriptionsStore subscriptionStore; private boolean allowAnonymous; private boolean allowZeroByteClientId; private IAuthorizator m_authorizator; @@ -141,7 +142,6 @@ public class ProtocolProcessor { private IAuthenticator m_authenticator; private BrokerInterceptor m_interceptor; - private String m_server_port; private Qos0PublishHandler qos0PublishHandler; private Qos1PublishHandler qos1PublishHandler; @@ -155,23 +155,23 @@ public class ProtocolProcessor { ProtocolProcessor() { } - public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, - IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator, - BrokerInterceptor interceptor) { + public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, + IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator, + BrokerInterceptor interceptor) { init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator, interceptor, null); } - public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, - IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, - IAuthorizator authorizator, BrokerInterceptor interceptor) { + public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, + IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, + IAuthorizator authorizator, BrokerInterceptor interceptor) { init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, null); } - public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, - IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, - IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { + public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, + IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, + IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { init(new ConnectionDescriptorStore(sessionsStore), subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, serverPort); } @@ -194,7 +194,7 @@ public class ProtocolProcessor { * @param interceptor * to notify events to an intercept handler */ - void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsStore subscriptions, + void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { @@ -212,7 +212,7 @@ public class ProtocolProcessor { m_authenticator = authenticator; m_messagesStore = storageService; m_sessionsStore = sessionsStore; - m_server_port = serverPort; + subscriptionStore = sessionsStore.subscriptionStore(); LOG.info("Initializing messages publisher..."); final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors); @@ -468,12 +468,11 @@ public class ProtocolProcessor { LOG.trace("retrieving inflight for messageID <{}>", messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - StoredMessage inflightMsg = targetSession.getInflightMessage(messageID); - targetSession.inFlightAcknowledged(messageID); + StoredMessage inflightMsg = targetSession.inFlightAcknowledged(messageID); String topic = inflightMsg.getTopic(); - m_interceptor - .notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID)); + InterceptAcknowledgedMessage wrapped = new InterceptAcknowledgedMessage(inflightMsg, topic, username, messageID); + m_interceptor.notifyMessageAcknowledged(wrapped); } public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) { @@ -539,9 +538,9 @@ public class ProtocolProcessor { } else { toStoreMsg.setClientID(clientId); } - if (qos == EXACTLY_ONCE) { // QoS2 - guid = m_messagesStore.storePublishForFuture(toStoreMsg); - } +// if (qos == EXACTLY_ONCE) { // QoS2 +// guid = m_messagesStore.storePublishForFuture(toStoreMsg); +// } this.messagesPublisher.publish2Subscribers(toStoreMsg, topic); if (!msg.fixedHeader().isRetain()) { @@ -552,11 +551,11 @@ public class ProtocolProcessor { m_messagesStore.cleanRetained(topic); return; } - if (guid == null) { - // before wasn't stored - guid = m_messagesStore.storePublishForFuture(toStoreMsg); - } - m_messagesStore.storeRetained(topic, guid); +// if (guid == null) { +// // before wasn't stored +// guid = m_messagesStore.storePublishForFuture(toStoreMsg); +// } + m_messagesStore.storeRetained(topic, toStoreMsg); } /** @@ -597,7 +596,8 @@ public class ProtocolProcessor { ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); // remove from the inflight and move to the QoS2 second phase queue int messageID = messageId(msg); - targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); + StoredMessage ackedMsg = targetSession.inFlightAcknowledged(messageID); + targetSession.moveInFlightToSecondPhaseAckWaiting(messageID, ackedMsg); // once received a PUBREC reply with a PUBREL(messageID) LOG.debug("Processing PUBREC message. CId={}, messageId={}", clientID, messageID); @@ -678,7 +678,7 @@ public class ProtocolProcessor { if (descriptor.cleanSession) { LOG.info("Removing saved subscriptions. CId={}", descriptor.clientID); - m_sessionsStore.wipeSubscriptions(clientID); + subscriptionStore.wipeSubscriptions(clientID); LOG.info("The saved subscriptions have been removed. CId={}", descriptor.clientID); } return true; @@ -918,9 +918,8 @@ public class ProtocolProcessor { emptyQueue = true; } else { // recreate a publish from stored publish in queue - boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null; MqttPublishMessage pubMsg = createPublishForQos( msg.getTopic(), msg.getQos(), msg.getPayload(), - retained, 0); + msg.isRetained(), 0); channel.write(pubMsg); } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 72159b5f..ae7cf9ed 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -25,17 +25,15 @@ import io.moquette.server.config.IResourceLoader; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.IStore; -import io.moquette.spi.impl.security.ACLFileParser; -import io.moquette.spi.impl.security.AcceptAllAuthenticator; -import io.moquette.spi.impl.security.DenyAllAuthorizator; -import io.moquette.spi.impl.security.PermitAllAuthorizator; -import io.moquette.spi.impl.security.ResourceAuthenticator; +import io.moquette.spi.ISubscriptionsStore; +import io.moquette.spi.impl.security.*; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.text.ParseException; @@ -50,16 +48,13 @@ public class ProtocolProcessorBootstrapper { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessorBootstrapper.class); public static final String MAPDB_STORE_CLASS = "io.moquette.persistence.MemoryStorageService"; - private SubscriptionsStore subscriptions; - private ISessionsStore m_sessionsStore; + private ISubscriptionsStore subscriptionsStore; + private Runnable storeShutdown; - private BrokerInterceptor m_interceptor; - private final ProtocolProcessor m_processor = new ProtocolProcessor(); - private ConnectionDescriptorStore connectionDescriptors; public ProtocolProcessorBootstrapper() { @@ -86,8 +81,6 @@ public class ProtocolProcessorBootstrapper { */ public ProtocolProcessor init(IConfig props, List embeddedObservers, IAuthenticator authenticator, IAuthorizator authorizator, Server server) { - subscriptions = new SubscriptionsStore(); - IMessagesStore messagesStore; LOG.info("Initializing messages and sessions stores..."); String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, MAPDB_STORE_CLASS); @@ -98,6 +91,7 @@ public class ProtocolProcessorBootstrapper { final IStore store = loadClass(storageClassName, IStore.class, Server.class, server); messagesStore = store.messagesStore(); m_sessionsStore = store.sessionsStore(); + this.subscriptionsStore = m_sessionsStore.subscriptionStore(); storeShutdown = new Runnable() { @Override @@ -116,9 +110,10 @@ public class ProtocolProcessorBootstrapper { observers.add(handler); } } - m_interceptor = new BrokerInterceptor(props, observers); + BrokerInterceptor interceptor = new BrokerInterceptor(props, observers); LOG.info("Initializing subscriptions store..."); + SubscriptionsDirectory subscriptions = new SubscriptionsDirectory(); subscriptions.init(m_sessionsStore); LOG.info("Configuring MQTT authenticator..."); @@ -170,7 +165,7 @@ public class ProtocolProcessorBootstrapper { boolean allowZeroByteClientId = Boolean .parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false")); m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator, - allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, + allowAnonymous, allowZeroByteClientId, authorizator, interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); return m_processor; } @@ -262,7 +257,7 @@ public class ProtocolProcessorBootstrapper { } public List getSubscriptions() { - return m_sessionsStore.getSubscriptions(); + return this.subscriptionsStore.getSubscriptions(); } public void shutdown() { diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 1132d049..9bd8c34f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -19,9 +19,6 @@ package io.moquette.spi.impl; import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; -import io.moquette.spi.MessageGUID; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.security.IAuthorizator; import io.netty.channel.Channel; @@ -31,7 +28,7 @@ import io.netty.handler.codec.mqtt.MqttPubAckMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; + import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; @@ -79,11 +76,7 @@ class Qos1PublishHandler extends QosPublishHandler { m_messagesStore.cleanRetained(topic); } else { // before wasn't stored - MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); - - //TODO Il messaggio non ha il guid settato -// m_messagesStore.storeRetained(topic, toStoreMsg.getGuid()); - m_messagesStore.storeRetained(topic, guid); + m_messagesStore.storeRetained(topic, toStoreMsg); } } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index 37021b22..dd1cc6a7 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -21,16 +21,13 @@ import io.moquette.server.netty.NettyUtils; import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.MessageGUID; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.security.IAuthorizator; import io.netty.channel.Channel; import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; import static io.moquette.spi.impl.DebugUtils.payload2Str; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; @@ -42,14 +39,14 @@ class Qos2PublishHandler extends QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos2PublishHandler.class); - private final SubscriptionsStore subscriptions; + private final SubscriptionsDirectory subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; private final ConnectionDescriptorStore connectionDescriptors; private final ISessionsStore m_sessionsStore; private final MessagesPublisher publisher; - public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, + public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsDirectory subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, MessagesPublisher messagesPublisher) { @@ -81,23 +78,22 @@ class Qos2PublishHandler extends QosPublishHandler { LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getPayload()), subscriptions.dumpTree()); } - // QoS2 - MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); - ClientSession sourceSession = m_sessionsStore.sessionForClient(clientID); - sourceSession.markAsInboundInflight(messageID, guid); + m_sessionsStore.sessionForClient(clientID) + .markAsInboundInflight(messageID, toStoreMsg); sendPubRec(clientID, messageID); // Next the client will send us a pub rel // NB publish to subscribers for QoS 2 happen upon PUBREL from publisher - if (msg.fixedHeader().isRetain()) { - if (msg.payload().readableBytes() == 0) { - m_messagesStore.cleanRetained(topic); - } else { - m_messagesStore.storeRetained(topic, guid); - } - } +// if (msg.fixedHeader().isRetain()) { +// if (msg.payload().readableBytes() == 0) { +// m_messagesStore.cleanRetained(topic); +// } else { +// m_messagesStore.storeRetained(topic, toStoreMsg); +// } +// } + //TODO this should happen on PUB_REL, else we notify false positive m_interceptor.notifyTopicPublished(msg, clientID, username); } @@ -123,10 +119,13 @@ class Qos2PublishHandler extends QosPublishHandler { if (evt.getPayload().readableBytes() == 0) { m_messagesStore.cleanRetained(topic); } else { - m_messagesStore.storeRetained(topic, evt.getGuid()); + m_messagesStore.storeRetained(topic, evt); } } + //TODO here we should notify to the listeners + //m_interceptor.notifyTopicPublished(msg, clientID, username); + sendPubComp(clientID, messageID); } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java index 606b13a0..37f7392a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java @@ -16,7 +16,7 @@ package io.moquette.spi.impl.subscriptions; -import io.moquette.spi.ISessionsStore.ClientTopicCouple; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import io.netty.handler.codec.mqtt.MqttQoS; import java.io.Serializable; diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java similarity index 91% rename from broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java rename to broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java index 813d9a2e..33228f92 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java @@ -17,7 +17,8 @@ package io.moquette.spi.impl.subscriptions; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.ISessionsStore.ClientTopicCouple; +import io.moquette.spi.ISubscriptionsStore; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; @@ -27,7 +28,7 @@ import java.util.concurrent.atomic.AtomicReference; /** * Represents a tree of topics subscriptions. */ -public class SubscriptionsStore { +public class SubscriptionsDirectory { public static class NodeCouple { @@ -79,8 +80,9 @@ public class SubscriptionsStore { } private AtomicReference subscriptions = new AtomicReference<>(new TreeNode()); - private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); + private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsDirectory.class); private volatile ISessionsStore m_sessionsStore; + private volatile ISubscriptionsStore subscriptionsStore; /** * Initialize the subscription tree with the list of subscriptions. Maintained for compatibility @@ -92,21 +94,20 @@ public class SubscriptionsStore { public void init(ISessionsStore sessionsStore) { LOG.info("Initializing subscriptions store..."); m_sessionsStore = sessionsStore; - List subscriptions = sessionsStore.listAllSubscriptions(); + this.subscriptionsStore = sessionsStore.subscriptionStore(); + List subscriptions = this.subscriptionsStore.listAllSubscriptions(); // reload any subscriptions persisted if (LOG.isTraceEnabled()) { - LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}.", dumpTree()); + LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}", dumpTree()); } for (ClientTopicCouple clientTopic : subscriptions) { - LOG.info( - "Re-subscribing client to topic. ClientId = {}, topicFilter = {}.", - clientTopic.clientID, - clientTopic.topicFilter); + LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", clientTopic.clientID, + clientTopic.topicFilter); add(clientTopic); } if (LOG.isTraceEnabled()) { - LOG.trace("The stored subscriptions have been reloaded. SubscriptionTree = {}.", dumpTree()); + LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree()); } } @@ -204,7 +205,7 @@ public class SubscriptionsStore { Map subsForClient = new HashMap<>(); for (ClientTopicCouple matchingCouple : matchingSubs) { Subscription existingSub = subsForClient.get(matchingCouple.clientID); - Subscription sub = m_sessionsStore.getSubscription(matchingCouple); + Subscription sub = this.subscriptionsStore.getSubscription(matchingCouple); if (sub == null) { // if the m_sessionStore hasn't the sub because the client disconnected continue; diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java index c956ab42..764264c9 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java @@ -16,7 +16,7 @@ package io.moquette.spi.impl.subscriptions; -import io.moquette.spi.ISessionsStore.ClientTopicCouple; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import java.util.*; import java.util.concurrent.LinkedBlockingQueue; diff --git a/broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java b/broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java new file mode 100644 index 00000000..322e7d99 --- /dev/null +++ b/broker/src/test/java/io/moquette/persistence/MessageStoreTCK.java @@ -0,0 +1,92 @@ +package io.moquette.persistence; + +import io.moquette.spi.*; +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; +import io.netty.buffer.ByteBuf; +import io.netty.handler.codec.mqtt.MqttQoS; +import org.junit.Test; + +import java.util.List; + +import static io.moquette.spi.impl.subscriptions.Topic.asTopic; +import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; +import static io.netty.handler.codec.mqtt.MqttQoS.EXACTLY_ONCE; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * Defines all test that an implementation of a IMessageStore should satisfy. + * */ +public abstract class MessageStoreTCK { + + public static final String TEST_CLIENT = "TestClient"; + protected ISessionsStore sessionsStore; + protected IMessagesStore messagesStore; + + @Test + public void testDropMessagesInSessionDoesntCleanAnyRetainedStoredMessages() { + final ClientSession session = sessionsStore.createNewSession(TEST_CLIENT, true); + StoredMessage publishToStore = new StoredMessage("Hello".getBytes(), EXACTLY_ONCE, "/topic"); + publishToStore.setClientID(TEST_CLIENT); + publishToStore.setRetained(true); + messagesStore.storeRetained(new Topic("/topic"), publishToStore); + + // Exercise + session.cleanSession(); + + // Verify the message store for session is empty. + StoredMessage storedPublish = messagesStore.searchMatching(new IMatchingCondition() { + + @Override + public boolean match(Topic key) { + return key.match(new Topic("/topic")); + } + }).iterator().next(); + assertNotNull("The stored retained message must be present after client's session drop", storedPublish); + } + + @Test + public void testStoreRetained() { + StoredMessage msgStored = new StoredMessage("Hello".getBytes(), MqttQoS.AT_LEAST_ONCE, "/topic"); + msgStored.setClientID(TEST_CLIENT); + + messagesStore.storeRetained(asTopic("/topic"), msgStored); + + //Verify the message is in the store + StoredMessage msgRetrieved = messagesStore.searchMatching(new IMatchingCondition() { + + @Override + public boolean match(Topic key) { + return key.match(new Topic("/topic")); + } + }).iterator().next(); + + final ByteBuf payload = msgRetrieved.getPayload(); + byte[] content = new byte[payload.readableBytes()]; + payload.readBytes(content); + assertEquals("Hello", new String(content)); + } + + @Test + public void givenSubscriptionAlreadyStoredIsOverwrittenByAnotherWithSameTopic() { + ClientSession session1 = sessionsStore.createNewSession("SESSION_ID_1", true); + + // Subscribe on /topic with QOSType.MOST_ONE + Subscription oldSubscription = new Subscription(session1.clientID, new Topic("/topic"), AT_MOST_ONCE); + session1.subscribe(oldSubscription); + + // Subscribe on /topic again that overrides the previous subscription. + Subscription overridingSubscription = new Subscription(session1.clientID, new Topic("/topic"), EXACTLY_ONCE); + session1.subscribe(overridingSubscription); + + // Verify + final ISubscriptionsStore subscriptionsStore = sessionsStore.subscriptionStore(); + List subscriptions = subscriptionsStore.listAllSubscriptions(); + assertEquals(1, subscriptions.size()); + Subscription sub = subscriptionsStore.getSubscription(subscriptions.get(0)); + assertEquals(overridingSubscription.getRequestedQos(), sub.getRequestedQos()); + } +} diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index 2cf8a2eb..c2be220c 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -18,7 +18,7 @@ package io.moquette.spi; import io.moquette.persistence.MemoryStorageService; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Topic; import static io.netty.handler.codec.mqtt.MqttQoS.*; import org.junit.Before; @@ -31,11 +31,11 @@ public class ClientSessionTest { ClientSession session1; ClientSession session2; ISessionsStore sessionsStore; - SubscriptionsStore store; + SubscriptionsDirectory store; @Before public void setUp() { - store = new SubscriptionsStore(); + store = new SubscriptionsDirectory(); MemoryStorageService storageService = new MemoryStorageService(); this.sessionsStore = storageService.sessionsStore(); store.init(sessionsStore); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 478feada..ca96cae7 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -26,7 +26,7 @@ import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.security.IAuthorizator; import io.netty.channel.embedded.EmbeddedChannel; @@ -62,7 +62,7 @@ public class ProtocolProcessorTest { IMessagesStore m_messagesStore; ISessionsStore m_sessionStore; - SubscriptionsStore subscriptions; + SubscriptionsDirectory subscriptions; MockAuthenticator m_mockAuthenticator; @Before @@ -92,7 +92,7 @@ public class ProtocolProcessorTest { users.put(TEST_USER, TEST_PWD); m_mockAuthenticator = new MockAuthenticator(clientIds, users); - subscriptions = new SubscriptionsStore(); + subscriptions = new SubscriptionsDirectory(); subscriptions.init(memStorage.sessionsStore()); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, @@ -104,7 +104,7 @@ public class ProtocolProcessorTest { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true - SubscriptionsStore subs = new SubscriptionsStore() { + SubscriptionsDirectory subs = new SubscriptionsDirectory() { @Override public List matches(Topic topic) { @@ -152,7 +152,7 @@ public class ProtocolProcessorTest { MqttQoS.AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true - SubscriptionsStore subs = new SubscriptionsStore() { + SubscriptionsDirectory subs = new SubscriptionsDirectory() { @Override public List matches(Topic topic) { @@ -311,7 +311,7 @@ public class ProtocolProcessorTest { new Subscription(FAKE_PUBLISHER_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true - SubscriptionsStore subs = new SubscriptionsStore() { + SubscriptionsDirectory subs = new SubscriptionsDirectory() { @Override public List matches(Topic topic) { @@ -355,14 +355,14 @@ public class ProtocolProcessorTest { @Test public void testRepublishAndConsumePersistedMessages_onReconnect() { - SubscriptionsStore subs = mock(SubscriptionsStore.class); + SubscriptionsDirectory subs = mock(SubscriptionsDirectory.class); List emptySubs = Collections.emptyList(); when(subs.matches(any(Topic.class))).thenReturn(emptySubs); StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(), MqttQoS.EXACTLY_ONCE, "/topic"); retainedMessage.setRetained(true); retainedMessage.setClientID(FAKE_PUBLISHER_ID); - m_messagesStore.storePublishForFuture(retainedMessage); + m_messagesStore.storeRetained(new Topic("/topic"), retainedMessage); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); @@ -382,7 +382,7 @@ public class ProtocolProcessorTest { // create an inactive session for Subscriber m_sessionStore.createNewSession("Subscriber", false); - SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); + SubscriptionsDirectory mockedSubscriptions = mock(SubscriptionsDirectory.class); Subscription inactiveSub = new Subscription("Subscriber", new Topic("/topic"), MqttQoS.AT_LEAST_ONCE); List inactiveSubscriptions = Collections.singletonList(inactiveSub); when(mockedSubscriptions.matches(eq(new Topic("/topic")))).thenReturn(inactiveSubscriptions); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index c5826ed0..c2e71452 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -24,7 +24,7 @@ import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Topic; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.MqttConnectMessage; @@ -51,7 +51,7 @@ public class ProtocolProcessor_CONNECT_Test { IMessagesStore m_messagesStore; ISessionsStore m_sessionStore; - SubscriptionsStore subscriptions; + SubscriptionsDirectory subscriptions; MockAuthenticator m_mockAuthenticator; @Before @@ -73,7 +73,7 @@ public class ProtocolProcessor_CONNECT_Test { users.put(ProtocolProcessorTest.TEST_USER, ProtocolProcessorTest.TEST_PWD); m_mockAuthenticator = new MockAuthenticator(clientIds, users); - subscriptions = new SubscriptionsStore(); + subscriptions = new SubscriptionsDirectory(); subscriptions.init(m_sessionStore); m_processor = new ProtocolProcessor(); m_processor.init( diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java similarity index 83% rename from broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java rename to broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java index c2352f27..83060fbd 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java @@ -16,42 +16,48 @@ package io.moquette.spi.impl.subscriptions; +import io.moquette.persistence.MemoryStorageService; import io.moquette.spi.ClientSession; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.ISessionsStore.ClientTopicCouple; -import io.moquette.persistence.MemoryStorageService; +import io.moquette.spi.ISubscriptionsStore; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import io.netty.handler.codec.mqtt.MqttQoS; import org.junit.Before; import org.junit.Test; + import java.io.IOException; import java.util.List; -import static org.assertj.core.api.Assertions.*; -public class SubscriptionsStoreTest { +import static org.assertj.core.api.Assertions.assertThat; - private SubscriptionsStore store; +public class SubscriptionsDirectoryTest { + + private SubscriptionsDirectory store; private ISessionsStore sessionsStore; + private ISubscriptionsStore subscriptionsStore; - public SubscriptionsStoreTest() { + public SubscriptionsDirectoryTest() { } @Before public void setUp() throws IOException { - store = new SubscriptionsStore(); + store = new SubscriptionsDirectory(); MemoryStorageService storageService = new MemoryStorageService(); this.sessionsStore = storageService.sessionsStore(); store.init(sessionsStore); + + this.subscriptionsStore = this.sessionsStore.subscriptionStore(); } @Test public void testMatchSimple() { Subscription slashSub = new Subscription("FAKE_CLI_ID_1", new Topic("/"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(slashSub); + this.subscriptionsStore.addNewSubscription(slashSub); store.add(slashSub.asClientTopicCouple()); assertThat(store.matches(new Topic("finance"))).isEmpty(); Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", new Topic("/finance"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(slashFinanceSub); + this.subscriptionsStore.addNewSubscription(slashFinanceSub); store.add(slashFinanceSub.asClientTopicCouple()); assertThat(store.matches(new Topic("finance"))).isEmpty(); @@ -62,12 +68,12 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti() { Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(anySub); + this.subscriptionsStore.addNewSubscription(anySub); store.add(anySub.asClientTopicCouple()); assertThat(store.matches(new Topic("finance"))).contains(anySub); Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", new Topic("finance/#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(financeAnySub); + this.subscriptionsStore.addNewSubscription(financeAnySub); store.add(financeAnySub.asClientTopicCouple()); assertThat(store.matches(new Topic("finance"))).containsExactlyInAnyOrder(financeAnySub, anySub); } @@ -76,8 +82,8 @@ public class SubscriptionsStoreTest { public void testMatchingDeepMulti_one_layer() { Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("#"), MqttQoS.AT_MOST_ONCE); Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", new Topic("finance/#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(anySub); - sessionsStore.addNewSubscription(financeAnySub); + this.subscriptionsStore.addNewSubscription(anySub); + this.subscriptionsStore.addNewSubscription(financeAnySub); store.add(anySub.asClientTopicCouple()); store.add(financeAnySub.asClientTopicCouple()); @@ -92,7 +98,7 @@ public class SubscriptionsStoreTest { "FAKE_CLI_ID_1", new Topic("finance/stock/#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(financeAnySub); + this.subscriptionsStore.addNewSubscription(financeAnySub); store.add(financeAnySub.asClientTopicCouple()); // Verify @@ -102,12 +108,12 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleSingle() { Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("+"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(anySub); + this.subscriptionsStore.addNewSubscription(anySub); store.add(anySub.asClientTopicCouple()); assertThat(store.matches(new Topic("finance"))).containsExactly(anySub); Subscription financeOne = new Subscription("FAKE_CLI_ID_1", new Topic("finance/+"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(financeOne); + this.subscriptionsStore.addNewSubscription(financeOne); store.add(financeOne.asClientTopicCouple()); assertThat(store.matches(new Topic("finance/stock"))).containsExactly(financeOne); } @@ -115,7 +121,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchManySingle() { Subscription manySub = new Subscription("FAKE_CLI_ID_1", new Topic("+/+"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(manySub); + this.subscriptionsStore.addNewSubscription(manySub); store.add(manySub.asClientTopicCouple()); // verify @@ -125,10 +131,10 @@ public class SubscriptionsStoreTest { @Test public void testMatchSlashSingle() { Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", new Topic("/+"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(slashPlusSub); + this.subscriptionsStore.addNewSubscription(slashPlusSub); store.add(slashPlusSub.asClientTopicCouple()); Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("+"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(anySub); + this.subscriptionsStore.addNewSubscription(anySub); store.add(anySub.asClientTopicCouple()); // Verify @@ -142,14 +148,14 @@ public class SubscriptionsStoreTest { "FAKE_CLI_ID_1", new Topic("/finance/+/ibm"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(slashPlusSub); + this.subscriptionsStore.addNewSubscription(slashPlusSub); store.add(slashPlusSub.asClientTopicCouple()); Subscription slashPlusDeepSub = new Subscription( "FAKE_CLI_ID_2", new Topic("/+/stock/+"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(slashPlusDeepSub); + this.subscriptionsStore.addNewSubscription(slashPlusDeepSub); store.add(slashPlusDeepSub.asClientTopicCouple()); // Verify @@ -160,7 +166,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_allTheTree() { Subscription sub = new Subscription("FAKE_CLI_ID_1", new Topic("#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(sub); + this.subscriptionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertThat(store.matches(new Topic("finance"))).isNotEmpty(); assertThat(store.matches(new Topic("finance/ibm"))).isNotEmpty(); @@ -170,7 +176,7 @@ public class SubscriptionsStoreTest { public void testMatchSimpleMulti_zeroLevel() { // check MULTI in case of zero level match Subscription sub = new Subscription("FAKE_CLI_ID_1", new Topic("finance/#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(sub); + this.subscriptionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertThat(store.matches(new Topic("finance"))).isNotEmpty(); } @@ -202,12 +208,12 @@ public class SubscriptionsStoreTest { private void assertMatch(String s, String t) { Topic subscription = new Topic(s); Topic topic = new Topic(t); - store = new SubscriptionsStore(); + store = new SubscriptionsDirectory(); MemoryStorageService memStore = new MemoryStorageService(); ISessionsStore aSessionsStore = memStore.sessionsStore(); store.init(aSessionsStore); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, MqttQoS.AT_MOST_ONCE); - aSessionsStore.addNewSubscription(sub); + aSessionsStore.subscriptionStore().addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertThat(store.matches(topic)).isNotEmpty(); } @@ -215,11 +221,11 @@ public class SubscriptionsStoreTest { private void assertNotMatch(String s, String t) { Topic subscription = new Topic(s); Topic topic = new Topic(t); - store = new SubscriptionsStore(); + store = new SubscriptionsDirectory(); MemoryStorageService memStore = new MemoryStorageService(); store.init(memStore.sessionsStore()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(sub); + this.subscriptionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertThat(store.matches(topic)).isEmpty(); } @@ -228,7 +234,7 @@ public class SubscriptionsStoreTest { public void testRemoveClientSubscriptions_existingClientID() { String cliendID = "FAKE_CLID_1"; Subscription sub = new Subscription(cliendID, new Topic("finance/#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(sub); + this.subscriptionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); // Exercise @@ -242,7 +248,7 @@ public class SubscriptionsStoreTest { public void testRemoveClientSubscriptions_notexistingClientID() { String clientID = "FAKE_CLID_1"; Subscription s = new Subscription(clientID, new Topic("finance/#"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(s); + this.subscriptionsStore.addNewSubscription(s); store.add(s.asClientTopicCouple()); // Exercise @@ -255,10 +261,10 @@ public class SubscriptionsStoreTest { @Test public void testOverlappingSubscriptions() { Subscription genericSub = new Subscription("FAKE_CLI_ID_1", new Topic("a/+"), MqttQoS.EXACTLY_ONCE); - sessionsStore.addNewSubscription(genericSub); + this.subscriptionsStore.addNewSubscription(genericSub); store.add(genericSub.asClientTopicCouple()); Subscription specificSub = new Subscription("FAKE_CLI_ID_1", new Topic("a/b"), MqttQoS.AT_LEAST_ONCE); - sessionsStore.addNewSubscription(specificSub); + this.subscriptionsStore.addNewSubscription(specificSub); store.add(specificSub.asClientTopicCouple()); // Verify @@ -267,19 +273,19 @@ public class SubscriptionsStoreTest { @Test public void removeSubscription_withDifferentClients_subscribedSameTopic() { - SubscriptionsStore aStore = new SubscriptionsStore(); + SubscriptionsDirectory aStore = new SubscriptionsDirectory(); MemoryStorageService memStore = new MemoryStorageService(); ISessionsStore sessionsStore = memStore.sessionsStore(); aStore.init(sessionsStore); // subscribe a not active clientID1 to /topic sessionsStore.createNewSession("FAKE_CLI_ID_1", true); Subscription slashSub = new Subscription("FAKE_CLI_ID_1", new Topic("/topic"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(slashSub); + sessionsStore.subscriptionStore().addNewSubscription(slashSub); aStore.add(slashSub.asClientTopicCouple()); // subscribe an active clientID2 to /topic Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", new Topic("/topic"), MqttQoS.AT_MOST_ONCE); - sessionsStore.addNewSubscription(slashSub2); + sessionsStore.subscriptionStore().addNewSubscription(slashSub2); aStore.add(new ClientTopicCouple("FAKE_CLI_ID_2", new Topic("/topic"))); // Exercise @@ -325,7 +331,7 @@ public class SubscriptionsStoreTest { @Test public void testRecreatePath_emptyRoot() { TreeNode oldRoot = new TreeNode(); - final SubscriptionsStore.NodeCouple resp = store.recreatePath(new Topic("/finance"), oldRoot); + final SubscriptionsDirectory.NodeCouple resp = store.recreatePath(new Topic("/finance"), oldRoot); // Verify assertThat(resp.root).isNotNull(); @@ -337,8 +343,8 @@ public class SubscriptionsStoreTest { @Test public void testRecreatePath_1layer_tree() { TreeNode oldRoot = new TreeNode(); - final SubscriptionsStore.NodeCouple respFinance = store.recreatePath(new Topic("/finance"), oldRoot); - final SubscriptionsStore.NodeCouple respPlus = store.recreatePath(new Topic("/+"), respFinance.root); + final SubscriptionsDirectory.NodeCouple respFinance = store.recreatePath(new Topic("/finance"), oldRoot); + final SubscriptionsDirectory.NodeCouple respPlus = store.recreatePath(new Topic("/+"), respFinance.root); // Verify assertThat(respPlus.root).isNotNull(); diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TreeNodeTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TreeNodeTest.java index 431997ba..f3e95f9f 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TreeNodeTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TreeNodeTest.java @@ -17,6 +17,8 @@ package io.moquette.spi.impl.subscriptions; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISubscriptionsStore; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import org.junit.Test; import static org.junit.Assert.*; @@ -28,7 +30,7 @@ public class TreeNodeTest { TreeNode root = new TreeNode(); TreeNode subNode1 = new TreeNode(); - subNode1.addSubscription(new ISessionsStore.ClientTopicCouple("Client1", new Topic("/topic1"))); + subNode1.addSubscription(new ClientTopicCouple("Client1", new Topic("/topic1"))); root.addChild(subNode1); TreeNode updatedRoot = root.removeClientSubscriptions("Client1"); @@ -42,7 +44,7 @@ public class TreeNodeTest { TreeNode intermediateEmptyNode = new TreeNode(); TreeNode subNode1 = new TreeNode(); - subNode1.addSubscription(new ISessionsStore.ClientTopicCouple("Client1", new Topic("/sub/topic1"))); + subNode1.addSubscription(new ClientTopicCouple("Client1", new Topic("/sub/topic1"))); intermediateEmptyNode.addChild(subNode1); root.addChild(intermediateEmptyNode); @@ -57,11 +59,11 @@ public class TreeNodeTest { TreeNode intermediateEmptyNode = new TreeNode(); TreeNode subNode1 = new TreeNode(); - subNode1.addSubscription(new ISessionsStore.ClientTopicCouple("Client1", new Topic("/sub/topic1"))); + subNode1.addSubscription(new ClientTopicCouple("Client1", new Topic("/sub/topic1"))); intermediateEmptyNode.addChild(subNode1); TreeNode subNode2 = new TreeNode(); - subNode2.addSubscription(new ISessionsStore.ClientTopicCouple("Client2", new Topic("/sub/topic2"))); + subNode2.addSubscription(new ClientTopicCouple("Client2", new Topic("/sub/topic2"))); intermediateEmptyNode.addChild(subNode2); root.addChild(intermediateEmptyNode); diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 69c99512..f1b2907a 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -9,7 +9,7 @@ log4j.logger.io.moquette.spi.impl.ProtocolProcessor=WARN log4j.logger.io.moquette.server.netty.NettyMQTTHandler=WARN #Storage service -#log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG +#log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsDirectorytory=DEBUG log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=WARN diff --git a/distribution/src/main/resources/log4j.properties b/distribution/src/main/resources/log4j.properties index 4b9ec64f..61764438 100644 --- a/distribution/src/main/resources/log4j.properties +++ b/distribution/src/main/resources/log4j.properties @@ -11,7 +11,7 @@ log4j.logger.io.moquette.server.netty.NettyMQTTHandler=WARN #log4j.logger.io.moquette.server.netty.NettyMQTTHandler=DEBUG log4j.logger.io.moquette.parser.netty=WARN #Storage server -#log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG +#log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsDirectorytory=DEBUG #log4j.logger.io.moquette.spi.impl.HawtDBStorageService=DEBUG # stdout appender is set to be a ConsoleAppender. @@ -44,4 +44,4 @@ log4j.appender.messagelog.MaxBackupIndex=1 log4j.appender.messagelog.layout=org.apache.log4j.PatternLayout log4j.appender.messagelog.layout.ConversionPattern=%d{HH:mm:ss,SSS} [%t] %-5p %c{1} %L %x - %m%n -log4j.category.messageLogger=WARN, messagelog \ No newline at end of file +log4j.category.messageLogger=WARN, messagelog diff --git a/h2_storage/pom.xml b/h2_storage/pom.xml new file mode 100644 index 00000000..e13182a4 --- /dev/null +++ b/h2_storage/pom.xml @@ -0,0 +1,129 @@ + + 4.0.0 + + ../ + moquette-parent + io.moquette + 0.10-SNAPSHOT + + + moquette-h2-storage + jar + Moquette - H2 storage + + + UTF-8 + io.moquette.moquette-broker + + + + + ${project.groupId} + moquette-broker + ${project.version} + + + + org.slf4j + slf4j-api + 1.7.5 + + + + com.h2database + h2-mvstore + 1.4.192 + + + + + junit + junit + 4.8.2 + test + + + + org.slf4j + slf4j-log4j12 + 1.7.5 + test + + + + + org.mockito + mockito-core + 1.9.0 + test + + + + org.assertj + assertj-core + + 2.6.0 + test + + + + io.moquette + moquette-broker + test-jar + ${project.version} + test + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + **/*Test.java + + + + + + + maven-jar-plugin + + + ${project.build.outputDirectory}/META-INF/MANIFEST.MF + + + + + + test-jar + + + + + + org.apache.felix + maven-bundle-plugin + + + bundle-manifest + process-classes + + manifest + + + + + + + + + + bintray + https://api.bintray.com/maven/${bintray.repo}/${bintray.package} + + + diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java new file mode 100644 index 00000000..776c4a29 --- /dev/null +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java @@ -0,0 +1,93 @@ +/* + * Copyright (c) 2012-2017 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.persistence.h2; + +import io.moquette.spi.IMatchingCondition; +import io.moquette.spi.IMessagesStore; +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.MessageGUID; +import io.moquette.spi.impl.subscriptions.Topic; +import org.h2.mvstore.Cursor; +import org.h2.mvstore.MVMap; +import org.h2.mvstore.MVStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sun.plugin2.message.Message; + +import java.util.*; + +class H2MessagesStore implements IMessagesStore { + + private static final Logger LOG = LoggerFactory.getLogger(H2MessagesStore.class); + + private final MVStore mvStore; + + // maps clientID -> guid + private MVMap retainedStore; + // maps guid to message, it's message store + private MVMap persistentMessageStore; + + private MVMap new_retainedStore; + + public H2MessagesStore(MVStore mvStore) { + this.mvStore = mvStore; + } + + @Override + public void initStore() { + retainedStore = mvStore.openMap("retained"); + persistentMessageStore = mvStore.openMap("oldPersistedMessages"); + new_retainedStore = mvStore.openMap("persistedMessages"); + LOG.info("Initialized message H2 store"); + } + + @Override + public void storeRetained(Topic topic, StoredMessage storedMessage) { + LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID()); + if (storedMessage.getClientID() == null) { + throw new IllegalArgumentException( "Message to be persisted must have a not null client ID"); + } + new_retainedStore.put(topic, storedMessage); + } + + @Override + public Collection searchMatching(IMatchingCondition condition) { + LOG.debug("Scanning retained messages"); + List results = new ArrayList<>(); + + Cursor mapCursor = retainedStore.cursor(null); + while (mapCursor.hasNext()) { + final MessageGUID guid = mapCursor.getValue(); + final Topic topic = mapCursor.getKey(); + StoredMessage storedMsg = persistentMessageStore.get(guid); + if (condition.match(topic)) { + results.add(storedMsg); + } + } + + if (LOG.isTraceEnabled()) { + LOG.trace("Retained messages have been scanned matchingMessages={}", results); + } + + return results; + } + + @Override + public void cleanRetained(Topic topic) { + LOG.debug("Cleaning retained messages. Topic={}", topic); + retainedStore.remove(topic); + } +} diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java new file mode 100644 index 00000000..3f661e4c --- /dev/null +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java @@ -0,0 +1,83 @@ +package io.moquette.persistence.h2; + +import org.h2.mvstore.MVMap; +import org.h2.mvstore.MVStore; + +import java.util.AbstractQueue; +import java.util.Iterator; +import java.util.concurrent.atomic.AtomicLong; + +class H2PersistentQueue extends AbstractQueue { + + private final MVMap mvMap; + private final MVMap metadataMap; + private final AtomicLong head; + private final AtomicLong tail; + + H2PersistentQueue(MVStore store, String queueName) { + if (queueName == null || queueName.isEmpty()) { + throw new IllegalArgumentException("queueName parameter can't be empty or null"); + } + this.mvMap = store.openMap("queue_" + queueName); + this.metadataMap = store.openMap("queue_" + queueName + "_meta"); + + //setup head index + long headIdx = 0L; + if (this.metadataMap.containsKey("head")) { + headIdx = this.metadataMap.get("head"); + } else { + this.metadataMap.put("head", headIdx); + } + this.head = new AtomicLong(headIdx); + + //setup tail index + long tailIdx = 0L; + if (this.metadataMap.containsKey("tail")) { + tailIdx = this.metadataMap.get("tail"); + } else { + this.metadataMap.put("tail", tailIdx); + } + this.tail = new AtomicLong(tailIdx); + } + + @Override + public Iterator iterator() { + return null; + } + + @Override + public int size() { + return this.head.intValue() - this.tail.intValue(); + } + + @Override + public boolean offer(T t) { + if (t == null) { + throw new NullPointerException("Inserted element can't be null"); + } + final long nextHead = head.getAndIncrement(); + this.mvMap.put(nextHead, t); + this.metadataMap.put("head", nextHead + 1); + return true; + } + + @Override + public T poll() { + if (head.equals(tail)) { + return null; + } + final long nextTail = tail.getAndIncrement(); + final T tail = this.mvMap.get(nextTail); + this.metadataMap.put("tail", nextTail + 1); + return tail; + } + + @Override + public T peek() { + if (head.equals(tail)) { + return null; + } + return this.mvMap.get(tail.get()); + } + +} diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java new file mode 100644 index 00000000..43b54a19 --- /dev/null +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentStore.java @@ -0,0 +1,94 @@ +/* + * Copyright (c) 2012-2017 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.persistence.h2; + +import io.moquette.BrokerConstants; +import io.moquette.server.config.IConfig; +import io.moquette.spi.IMessagesStore; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.IStore; +import org.h2.mvstore.MVStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +public class H2PersistentStore implements IStore { + + private static final Logger LOG = LoggerFactory.getLogger(H2PersistentStore.class); + + private final String storePath; + private final int autosaveInterval; // in seconds + protected final ScheduledExecutorService scheduler; + private MVStore mvStore; + + private IMessagesStore messageStore; + private ISessionsStore sessionsStore; + + public H2PersistentStore(IConfig props, ScheduledExecutorService scheduler) { + this.storePath = props.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, ""); + this.autosaveInterval = Integer.parseInt(props.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30")); + this.scheduler = scheduler; + } + + @Override + public void initStore() { + LOG.info("Initializing H2 store"); + if (storePath == null || storePath.isEmpty()) { + LOG.warn("H2 store file path is empty, using in-memory store"); + mvStore = MVStore.open(null); + + } else { + mvStore = new MVStore.Builder() + .fileName(storePath) + .autoCommitDisabled() + .open(); + } + + LOG.info("Scheduling H2 commit task"); + scheduler.scheduleWithFixedDelay(new Runnable() { + + @Override + public void run() { + LOG.debug("Committing to H2"); + mvStore.commit(); + } + }, this.autosaveInterval, this.autosaveInterval, TimeUnit.SECONDS); + + messageStore = new H2MessagesStore(mvStore); + messageStore.initStore(); + + sessionsStore = new H2SessionsStore(mvStore); + sessionsStore.initStore(); + } + + @Override + public void close() { + + } + + @Override + public IMessagesStore messagesStore() { + return messageStore; + } + + @Override + public ISessionsStore sessionsStore() { + return sessionsStore; + } + +} diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java new file mode 100644 index 00000000..c51292c3 --- /dev/null +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java @@ -0,0 +1,370 @@ +/* + * Copyright (c) 2012-2017 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.persistence.h2; + +import io.moquette.persistence.PersistentSession; +import io.moquette.spi.ClientSession; +import io.moquette.spi.IMessagesStore; +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISubscriptionsStore; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.Topic; +import org.h2.mvstore.Cursor; +import org.h2.mvstore.MVMap; +import org.h2.mvstore.MVStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { + + private static final Logger LOG = LoggerFactory.getLogger(H2SessionsStore.class); + + private final MVStore mvStore; + private ConcurrentMap sessions; + // maps clientID->[MessageId -> msg] + private ConcurrentMap> outboundFlightMessages; + // map clientID <-> set of currently in flight packet identifiers + private Map> inFlightIds; + // maps clientID->[MessageId -> guid] + private ConcurrentMap> secondPhaseStore; + + public H2SessionsStore(MVStore mvStore) { + this.mvStore = mvStore; + } + + @Override + public void initStore() { + this.sessions = mvStore.openMap("sessions"); + this.outboundFlightMessages = mvStore.openMap("outboundFlight"); + this.inFlightIds = mvStore.openMap("inflightPacketIDs"); + this.secondPhaseStore = mvStore.openMap("secondPhase"); + LOG.info("Initialized sessions H2 store"); + } + + @Override + public ISubscriptionsStore subscriptionStore() { + return this; + } + + @Override + public void updateCleanStatus(String clientID, boolean cleanSession) { + LOG.info("Updating cleanSession flag. CId={}, cleanSession={}", clientID, cleanSession); + this.sessions.put(clientID, new PersistentSession(cleanSession)); + } + + @Override + public void addNewSubscription(Subscription newSubscription) { + LOG.info("Adding new subscription CId={}, topics={}", newSubscription.getClientId(), + newSubscription.getTopicFilter()); + final String clientID = newSubscription.getClientId(); + final MVMap sessionSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + sessionSubscriptions.put(newSubscription.getTopicFilter(), newSubscription); + + if (LOG.isTraceEnabled()) { + LOG.trace("Subscription has been added CId={}, topics={}, clientSubscriptions={}", + newSubscription.getClientId(), newSubscription.getTopicFilter(), sessionSubscriptions); + } + } + + @Override + public void removeSubscription(Topic topicFilter, String clientID) { + LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter); + if (!this.mvStore.hasMap("subscriptions_" + clientID)) { + return; + } + final MVMap sessionsSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + sessionsSubscriptions.remove(topicFilter); + if (LOG.isDebugEnabled()) { + LOG.debug("Subscription has been removed. CId={}, topics={}, clientSubscriptions={}", clientID, + topicFilter, sessionsSubscriptions); + } + } + + @Override + public void wipeSubscriptions(String sessionID) { + LOG.info("Wiping subscriptions. CId={}", sessionID); + if (!this.mvStore.hasMap("subscriptions_" + sessionID)) { + return; + } + final MVMap subscriptions = this.mvStore.openMap("subscriptions_" + sessionID); + this.mvStore.removeMap(subscriptions); + if (LOG.isDebugEnabled()) { + LOG.debug("Subscriptions have been removed. CId={}, clientSubscriptions={}", sessionID, subscriptions); + } + } + + @Override + public List listAllSubscriptions() { + LOG.info("Retrieving existing subscriptions"); + final List allSubscriptions = new ArrayList<>(); + for (String clientID : this.sessions.keySet()) { + ConcurrentMap clientSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + for (Topic topicFilter : clientSubscriptions.keySet()) { + allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); + } + } + if (LOG.isTraceEnabled()) { + LOG.trace("The existing subscriptions have been retrieved. Result={}", allSubscriptions); + } + return allSubscriptions; + } + + @Override + public List getSubscriptions() { + LOG.debug("Retrieving existing subscriptions..."); + List subscriptions = new ArrayList<>(); + for (String clientID : this.sessions.keySet()) { + ConcurrentMap clientSubscriptions = this.mvStore.openMap("subscriptions_" + clientID); + subscriptions.addAll(clientSubscriptions.values()); + } + LOG.debug("Existing subscriptions has been retrieved Result={}", subscriptions); + return subscriptions; + } + + @Override + public Subscription getSubscription(ClientTopicCouple couple) { + ConcurrentMap clientSubscriptions = this.mvStore.openMap("subscriptions_" + couple.clientID); + LOG.debug("Retrieving subscriptions CId={}, subscriptions={}", couple.clientID, clientSubscriptions); + return clientSubscriptions.get(couple.topicFilter); + } + + @Override + public boolean contains(String clientID) { + return this.mvStore.hasMap("subscriptions_" + clientID); + } + + @Override + public ClientSession createNewSession(String clientID, boolean cleanSession) { + if (sessions.containsKey(clientID)) { + LOG.error("Unable to create a new session: the client ID is already in use. CId={}, cleanSession={}", + clientID, cleanSession); + throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); + } + LOG.debug("Creating new session. CId={}, cleanSession={}", clientID, cleanSession); + sessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); + return new ClientSession(clientID, this, this, cleanSession); + } + + @Override + public ClientSession sessionForClient(String clientID) { + LOG.debug("Retrieving session. CId={}", clientID); + if (!this.sessions.containsKey(clientID)) { + LOG.warn("Session does not exist. CId={}", clientID); + return null; + } + PersistentSession storedSession = this.sessions.get(clientID); + return new ClientSession(clientID, this, this, storedSession.cleanSession); + } + + @Override + public Collection getAllSessions() { + Collection result = new ArrayList<>(); + for (Map.Entry entry : this.sessions.entrySet()) { + result.add(new ClientSession(entry.getKey(), this, this, entry.getValue().cleanSession)); + } + return result; + } + + @Override + public StoredMessage inFlightAck(String clientID, int messageID) { + LOG.debug("Acknowledging inflight message CId={}, messageId={}", clientID, messageID); + ConcurrentMap m = this.outboundFlightMessages.get(clientID); + if (m == null) { + LOG.error("Can't find the inFlight record for client <{}>", clientID); + throw new RuntimeException("Can't find the inFlight record for client <" + clientID + ">"); + } + StoredMessage msg = m.remove(messageID); + this.outboundFlightMessages.put(clientID, m); + + // remove from the ids store + Set inFlightForClient = this.inFlightIds.get(clientID); + if (inFlightForClient != null) { + inFlightForClient.remove(messageID); + } + return msg; + } + + @Override + public void inFlight(String clientID, int messageID, StoredMessage msg) { + ConcurrentMap messages = this.outboundFlightMessages.get(clientID); + if (messages == null) { + messages = new ConcurrentHashMap<>(); + } + messages.put(messageID, msg); + this.outboundFlightMessages.put(clientID, messages); + } + + @Override + public int nextPacketID(String clientID) { + LOG.debug("Generating next packet ID CId={}", clientID); + Set inFlightForClient = this.inFlightIds.get(clientID); + if (inFlightForClient == null) { + int nextPacketId = 1; + inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap()); + inFlightForClient.add(nextPacketId); + this.inFlightIds.put(clientID, inFlightForClient); + return nextPacketId; + } + + int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); + int nextPacketId = (maxId % 0xFFFF) + 1; + inFlightForClient.add(nextPacketId); + LOG.debug("Next packet ID has been generated CId={}, result={}", clientID, nextPacketId); + return nextPacketId; + } + + + class H2BackedQueue extends AbstractQueue { + + private final MVMap map; + private final Cursor cursor; + + H2BackedQueue(MVMap map) { + this.cursor = map.cursor(null); + this.map = map; + } + + @Override + public Iterator iterator() { + return cursor; + } + + @Override + public int size() { + return map.size(); + } + + @Override + public boolean offer(T t) { + return false; + } + + @Override + public T poll() { + return null; + } + + @Override + public T peek() { + return null; + } + } + + //TODO + @Override + public BlockingQueue queue(String clientID) { + throw new RuntimeException("Not yet implemented"); + } + + //TODO + @Override + public void dropQueue(String clientID) { + throw new RuntimeException("Not yet implemented"); + } + + @Override + public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) { + LOG.debug("Moving inflight message to 2nd phase ack state CId={}, messageID={}", clientID, messageID); + ConcurrentMap m = this.secondPhaseStore.get(clientID); + if (m == null) { + String error = String.format("Can't find the inFlight record for client <%s> during the second phase of " + + "QoS2 pub", clientID); + LOG.error(error); + throw new RuntimeException(error); + } + m.put(messageID, msg); + this.outboundFlightMessages.put(clientID, m); + } + + @Override + public StoredMessage secondPhaseAcknowledged(String clientID, int messageID) { + LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); + final ConcurrentMap m = this.secondPhaseStore.get(clientID); + if (m == null) { + String error = String.format("Can't find the inFlight record for client <%s> during the second phase " + + "acking of QoS2 pub", clientID); + LOG.error(error); + throw new RuntimeException(error); + } + + StoredMessage msg = m.remove(messageID); + this.secondPhaseStore.put(clientID, m); + return msg; + } + + @Override + public int getInflightMessagesNo(String clientID) { + int totalInflight = 0; + ConcurrentMap inflightPerClient = this.mvStore.openMap(inboundStoreForClient(clientID)); + if (inflightPerClient != null) { + totalInflight += inflightPerClient.size(); + } + + Map secondPhaseInFlight = this.secondPhaseStore.get(clientID); + if (secondPhaseInFlight != null) { + totalInflight += secondPhaseInFlight.size(); + } + + Map outboundPerClient = outboundFlightMessages.get(clientID); + if (outboundPerClient != null) { + totalInflight += outboundPerClient.size(); + } + + return totalInflight; + } + + private static String inboundStoreForClient(String clientID) { + return "inboundInflight_" + clientID; + } + + @Override + public StoredMessage inboundInflight(String clientID, int messageID) { + LOG.debug("Mapping inbound message ID to GUID CId={}, messageId={}", clientID, messageID); + ConcurrentMap messageIdToGuid = this.mvStore.openMap(inboundStoreForClient(clientID)); + return messageIdToGuid.get(messageID); + } + + @Override + public void markAsInboundInflight(String clientID, int messageID, StoredMessage msg) { + ConcurrentMap messageIdToGuid = this.mvStore.openMap(inboundStoreForClient(clientID)); + messageIdToGuid.put(messageID, msg); + } + + @Override + public int getPendingPublishMessagesNo(String clientID) { + return queue(clientID).size(); + } + + @Override + public int getSecondPhaseAckPendingMessages(String clientID) { + if (!this.secondPhaseStore.containsKey(clientID)) + return 0; + return this.secondPhaseStore.get(clientID).size(); + } + + @Override + public void dropInFlightMessagesInSession(String clientID) { + this.secondPhaseStore.remove(clientID); + this.outboundFlightMessages.remove(clientID); + } + +} diff --git a/h2_storage/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java b/h2_storage/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java new file mode 100644 index 00000000..08ab010e --- /dev/null +++ b/h2_storage/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java @@ -0,0 +1,69 @@ +/* + * Copyright (c) 2012-2017 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.persistence.h2; + +import io.moquette.BrokerConstants; +import io.moquette.persistence.MessageStoreTCK; +import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; +import org.junit.After; +import org.junit.Before; + +import java.io.File; +import java.util.Properties; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; + +import static org.junit.Assert.assertFalse; + +public class H2MessagesStoreTest extends MessageStoreTCK { + + H2PersistentStore storageService; + + private ScheduledExecutorService scheduler; + + @Before + public void setUp() throws Exception { + scheduler = Executors.newScheduledThreadPool(1); + + cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); + Properties props = new Properties(); + props.setProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, BrokerConstants.DEFAULT_PERSISTENT_PATH); + IConfig conf = new MemoryConfig(props); + storageService = new H2PersistentStore(conf, scheduler); + storageService.initStore(); + messagesStore = storageService.messagesStore(); + sessionsStore = storageService.sessionsStore(); + } + + @After + public void tearDown() { + if (storageService != null) { + storageService.close(); + } + + scheduler.shutdown(); + cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); + } + + public static void cleanPersistenceFile(String fileName) { + File dbFile = new File(fileName); + if (dbFile.exists()) { + dbFile.delete(); + } + assertFalse(dbFile.exists()); + } +} diff --git a/h2_storage/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java b/h2_storage/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java new file mode 100644 index 00000000..4304b491 --- /dev/null +++ b/h2_storage/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java @@ -0,0 +1,96 @@ +package io.moquette.persistence.h2; + +import io.moquette.BrokerConstants; +import org.h2.mvstore.MVStore; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; + +import static org.junit.Assert.*; + +public class H2PersistentQueueTest { + + private MVStore mvStore; + + @Before + public void setUp() { + this.mvStore = new MVStore.Builder() + .fileName(BrokerConstants.DEFAULT_PERSISTENT_PATH) + .autoCommitDisabled() + .open(); + } + + @After + public void tearDown() { + File dbFile = new File(BrokerConstants.DEFAULT_PERSISTENT_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } + assertFalse(dbFile.exists()); + } + + @Test + public void testAdd() { + H2PersistentQueue sut = new H2PersistentQueue<>(this.mvStore, "test"); + + sut.add("Hello"); + sut.add("world"); + + assertEquals("Hello", sut.peek()); + assertEquals("Hello", sut.peek()); + assertEquals("peek just return elements, doesn't remove them", 2, sut.size()); + } + + @Test + public void testPoll() { + H2PersistentQueue sut = new H2PersistentQueue<>(this.mvStore, "test"); + sut.add("Hello"); + sut.add("world"); + + assertEquals("Hello", sut.poll()); + assertEquals("world", sut.poll()); + assertTrue("after poll 2 elements inserted before, should be empty", sut.isEmpty()); + } + + @Test + public void testPerformance() { + H2PersistentQueue sut = new H2PersistentQueue<>(this.mvStore, "test"); + + int numIterations = 10000000; + for (int i=0; i < numIterations; i++) { + sut.add("Hello"); + } + mvStore.commit(); + + for (int i=0; i < numIterations; i++) { + assertEquals("Hello", sut.poll()); + } + + assertTrue("should be empty", sut.isEmpty()); + } + + @Test + public void testReloadFromPersistedState() { + H2PersistentQueue before = new H2PersistentQueue<>(this.mvStore, "test"); + before.add("Hello"); + before.add("crazy"); + before.add("world"); + assertEquals("Hello", before.poll()); + this.mvStore.commit(); + this.mvStore.close(); + + this.mvStore = new MVStore.Builder() + .fileName(BrokerConstants.DEFAULT_PERSISTENT_PATH) + .autoCommitDisabled() + .open(); + + //now reload the persisted state + H2PersistentQueue after = new H2PersistentQueue<>(this.mvStore, "test"); + + assertEquals("crazy", after.poll()); + assertEquals("world", after.poll()); + assertTrue("should be empty", after.isEmpty()); + } +} diff --git a/mapdb_storage/pom.xml b/mapdb_storage/pom.xml index f276cffc..60094757 100644 --- a/mapdb_storage/pom.xml +++ b/mapdb_storage/pom.xml @@ -33,24 +33,9 @@ org.mapdb mapdb - - - 1.0.8 - - org.hdrhistogram - HdrHistogram - 2.1.4 - - - - com.hazelcast - hazelcast - 3.5.4 - - junit @@ -74,27 +59,6 @@ test - - - - - - - - - - - - - - - - - - - - - org.assertj assertj-core @@ -103,12 +67,13 @@ test - - - - - - + + io.moquette + moquette-broker + test-jar + ${project.version} + test + diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java index 82375454..548ced06 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java @@ -18,12 +18,15 @@ package io.moquette.persistence.mapdb; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; -import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.subscriptions.Topic; import org.mapdb.DB; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.*; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentMap; /** @@ -35,10 +38,7 @@ class MapDBMessagesStore implements IMessagesStore { private DB m_db; - // maps clientID -> guid - private ConcurrentMap m_retainedStore; - // maps guid to message, it's message store - private ConcurrentMap m_persistentMessageStore; + private ConcurrentMap m_retainedStore; MapDBMessagesStore(DB db) { m_db = db; @@ -47,66 +47,39 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void initStore() { m_retainedStore = m_db.getHashMap("retained"); - m_persistentMessageStore = m_db.getHashMap("persistedMessages"); LOG.info("Initialized store"); } - @Override - public void storeRetained(Topic topic, MessageGUID guid) { - LOG.debug("Storing retained messages. Topic={}, guid={}", topic, guid); - m_retainedStore.put(topic, guid); - } - @Override public Collection searchMatching(IMatchingCondition condition) { - LOG.debug("Scanning retained messages..."); + LOG.debug("Scanning retained messages"); List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - final MessageGUID guid = entry.getValue(); - StoredMessage storedMsg = m_persistentMessageStore.get(guid); + for (Map.Entry entry : m_retainedStore.entrySet()) { + StoredMessage storedMsg = entry.getValue(); if (condition.match(entry.getKey())) { results.add(storedMsg); } } if (LOG.isTraceEnabled()) { - LOG.trace("The retained messages have been scanned. MatchingMessages={}", results); + LOG.trace("Retained messages have been scanned matchingMessages={}", results); } return results; } - @Override - public MessageGUID storePublishForFuture(StoredMessage storedMessage) { - assert storedMessage.getClientID() != null : "Message to be persisted must have a valid client ID"; - MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); - storedMessage.setGuid(guid); - LOG.debug("Storing publish event. CId={}, guid={}, topic={}", storedMessage.getClientID(), guid, - storedMessage.getTopic()); - m_persistentMessageStore.put(guid, storedMessage); - return guid; - } - - @Override - public void dropInFlightMessagesInSession(Collection pendingAckMessages) { - //remove all guids from retained - Collection messagesToRemove = new HashSet<>(pendingAckMessages); - messagesToRemove.removeAll(m_retainedStore.values()); - - for (MessageGUID guid : messagesToRemove) { - m_persistentMessageStore.remove(guid); - } - } - - @Override - public StoredMessage getMessageByGuid(MessageGUID guid) { - LOG.debug("Retrieving stored message. Guid={}", guid); - return m_persistentMessageStore.get(guid); - } - @Override public void cleanRetained(Topic topic) { LOG.debug("Cleaning retained messages. Topic={}", topic); m_retainedStore.remove(topic); } + + @Override + public void storeRetained(Topic topic, StoredMessage storedMessage) { + LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID()); + if (storedMessage.getClientID() == null) { + throw new IllegalArgumentException( "Message to be persisted must have a not null client ID"); + } + m_retainedStore.put(topic, storedMessage); + } } diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java index 0a421998..0cb0b086 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java @@ -52,11 +52,6 @@ public class MapDBPersistentStore implements IStore { this.m_scheduler = scheduler; } - /** - * Factory method to create message store backed by MapDB - * - * @return the message store instance. - */ @Override public IMessagesStore messagesStore() { return m_messageStore; @@ -67,36 +62,32 @@ public class MapDBPersistentStore implements IStore { return m_sessionsStore; } + @Override public void initStore() { - LOG.info("Initializing MapDB store..."); + LOG.info("Initializing MapDB store"); if (m_storePath == null || m_storePath.isEmpty()) { - LOG.warn("The MapDB store file path is empty. Using in-memory store."); + LOG.warn("MapDB store file path is empty, using in-memory store"); m_db = DBMaker.newMemoryDB().make(); } else { File tmpFile; try { - LOG.info("Using user-defined MapDB store file. Path = {}.", m_storePath); + LOG.info("Using user-defined MapDB store file. Path={}", m_storePath); tmpFile = new File(m_storePath); boolean fileNewlyCreated = tmpFile.createNewFile(); - LOG.warn("Using {} MapDB store file. Path = {}.", fileNewlyCreated ? "fresh" : "existing", m_storePath); + LOG.warn("Using {} MapDB store file. Path={}", fileNewlyCreated ? "fresh" : "existing", m_storePath); } catch (IOException ex) { - LOG.error( - "Unable to open MapDB store file. Path = {}, cause = {}, errorMessage = {}.", - m_storePath, - ex.getCause(), - ex.getMessage()); - throw new RuntimeException( - "Can't create temp file for subscriptions storage [" + m_storePath + "]", - ex); + LOG.error("Unable to open MapDB store file. Path={}, cause={}, errorMessage={}", m_storePath, + ex.getCause(), ex.getMessage()); + throw new RuntimeException("Can't create temp subscriptions file storage [" + m_storePath + "]", ex); } m_db = DBMaker.newFileDB(tmpFile).make(); } - LOG.info("Scheduling MapDB commit task..."); + LOG.info("Scheduling MapDB commit task"); m_scheduler.scheduleWithFixedDelay(new Runnable() { @Override public void run() { - LOG.debug("Committing to MapDB..."); + LOG.debug("Committing to MapDB"); m_db.commit(); } }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS); @@ -105,21 +96,21 @@ public class MapDBPersistentStore implements IStore { m_messageStore = new MapDBMessagesStore(m_db); m_messageStore.initStore(); - m_sessionsStore = new MapDBSessionsStore(m_db, m_messageStore); + m_sessionsStore = new MapDBSessionsStore(m_db); m_sessionsStore.initStore(); } @Override public void close() { if (this.m_db.isClosed()) { - LOG.warn("The MapDB store is already closed. Nothing will be done."); + LOG.warn("MapDB store is already closed. Nothing will be done"); return; } - LOG.info("Performing last commit to MapDB..."); + LOG.info("Performing last commit to MapDB"); this.m_db.commit(); - LOG.info("Closing MapDB store..."); + LOG.info("Closing MapDB store"); this.m_db.close(); - LOG.info("Stopping MapDB commit tasks..."); + LOG.info("Stopping MapDB commit tasks"); //TODO th scheduler must be stopped by the owning (the instance of Server) //invalidate the added task @@ -129,9 +120,9 @@ public class MapDBPersistentStore implements IStore { } catch (InterruptedException e) { } if (!m_scheduler.isTerminated()) { - LOG.warn("Forcing shutdown of MapDB commit tasks..."); + LOG.warn("Forcing shutdown of MapDB commit tasks"); m_scheduler.shutdown(); } - LOG.info("The MapDB store has been closed successfully."); + LOG.info("MapDB store has been closed successfully"); } } diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java index f97922f7..501412d2 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java @@ -18,11 +18,9 @@ package io.moquette.persistence.mapdb; import io.moquette.persistence.PersistentSession; import io.moquette.spi.ClientSession; -import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.MessageGUID; -import io.moquette.spi.impl.Utils; +import io.moquette.spi.ISubscriptionsStore; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Topic; import org.mapdb.DB; @@ -37,34 +35,37 @@ import java.util.concurrent.ConcurrentMap; /** * ISessionsStore implementation backed by MapDB. */ -class MapDBSessionsStore implements ISessionsStore { +class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class); - // maps clientID->[MessageId -> guid] - private ConcurrentMap> m_inflightStore; + // maps clientID->[MessageId -> msg] + private ConcurrentMap> outboundFlightMessages; // map clientID <-> set of currently in flight packet identifiers private Map> m_inFlightIds; private ConcurrentMap m_persistentSessions; // maps clientID->[MessageId -> guid] - private ConcurrentMap> m_secondPhaseStore; + private ConcurrentMap> m_secondPhaseStore; private final DB m_db; - private final IMessagesStore m_messagesStore; - MapDBSessionsStore(DB db, IMessagesStore messagesStore) { + MapDBSessionsStore(DB db) { m_db = db; - m_messagesStore = messagesStore; } @Override public void initStore() { - m_inflightStore = m_db.getHashMap("inflight"); + outboundFlightMessages = m_db.getHashMap("outboundFlight"); m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); m_persistentSessions = m_db.getHashMap("sessions"); m_secondPhaseStore = m_db.getHashMap("secondPhase"); } + @Override + public ISubscriptionsStore subscriptionStore() { + return this; + } + @Override public void addNewSubscription(Subscription newSubscription) { LOG.info("Adding new subscription. ClientId={}, topics={}", newSubscription.getClientId(), @@ -94,7 +95,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void wipeSubscriptions(String clientID) { - LOG.info("Wiping subscriptions. ClientId={}", clientID); + LOG.info("Wiping subscriptions. CId={}", clientID); m_db.delete("subscriptions_" + clientID); if (LOG.isDebugEnabled()) { LOG.debug("Subscriptions have been removed. ClientId={}, clientSubscriptions={}", clientID, @@ -104,7 +105,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public List listAllSubscriptions() { - LOG.info("Retrieving existing subscriptions..."); + LOG.debug("Retrieving existing subscriptions"); final List allSubscriptions = new ArrayList<>(); for (String clientID : m_persistentSessions.keySet()) { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); @@ -121,19 +122,19 @@ class MapDBSessionsStore implements ISessionsStore { @Override public Subscription getSubscription(ClientTopicCouple couple) { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); - LOG.info("Retrieving subscriptions. ClientId={}, subscriptions={}", couple.clientID, clientSubscriptions); + LOG.debug("Retrieving subscriptions. CId={}, subscriptions={}", couple.clientID, clientSubscriptions); return clientSubscriptions.get(couple.topicFilter); } @Override public List getSubscriptions() { - LOG.info("Retrieving existing subscriptions..."); + LOG.debug("Retrieving existing subscriptions..."); List subscriptions = new ArrayList<>(); for (String clientID : m_persistentSessions.keySet()) { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); subscriptions.addAll(clientSubscriptions.values()); } - LOG.debug("The existing subscriptions have been retrieved. Result={}", subscriptions); + LOG.debug("Existing subscriptions has been retrieved Result={}", subscriptions); return subscriptions; } @@ -149,35 +150,35 @@ class MapDBSessionsStore implements ISessionsStore { clientID, cleanSession); throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); } - LOG.info("Creating new session. ClientId={}, cleanSession={}", clientID, cleanSession); + LOG.debug("Creating new session. CId={}, cleanSession={}", clientID, cleanSession); m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); - return new ClientSession(clientID, m_messagesStore, this, cleanSession); + return new ClientSession(clientID, this, this, cleanSession); } @Override public ClientSession sessionForClient(String clientID) { - LOG.info("Retrieving session. ClientId={}", clientID); + LOG.debug("Retrieving session CId={}", clientID); if (!m_persistentSessions.containsKey(clientID)) { - LOG.warn("The session does not exist. ClientId={}", clientID); + LOG.warn("Session does not exist CId={}", clientID); return null; } PersistentSession storedSession = m_persistentSessions.get(clientID); - return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); + return new ClientSession(clientID, this, this, storedSession.cleanSession); } @Override public Collection getAllSessions() { Collection result = new ArrayList<>(); for (Map.Entry entry : m_persistentSessions.entrySet()) { - result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); + result.add(new ClientSession(entry.getKey(), this, this, entry.getValue().cleanSession)); } return result; } @Override public void updateCleanStatus(String clientID, boolean cleanSession) { - LOG.info("Updating cleanSession flag. ClientId={}, cleanSession={}", clientID, cleanSession); + LOG.info("Updating cleanSession flag. CId={}, cleanSession={}", clientID, cleanSession); m_persistentSessions.put(clientID, new PersistentSession(cleanSession)); } @@ -186,9 +187,7 @@ class MapDBSessionsStore implements ISessionsStore { */ @Override public int nextPacketID(String clientID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Generating next packet ID. ClientId={}", clientID); - } + LOG.debug("Generating next packet ID CId={}", clientID); Set inFlightForClient = this.m_inFlightIds.get(clientID); if (inFlightForClient == null) { int nextPacketId = 1; @@ -202,39 +201,37 @@ class MapDBSessionsStore implements ISessionsStore { int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); int nextPacketId = (maxId % 0xFFFF) + 1; inFlightForClient.add(nextPacketId); - if (LOG.isDebugEnabled()) { - LOG.debug("The next packet ID has been generated. ClientId={}, result={}", clientID, nextPacketId); - } + LOG.debug("Next packet ID has been generated CId={}, result={}", clientID, nextPacketId); return nextPacketId; } @Override - public void inFlightAck(String clientID, int messageID) { - LOG.debug("Acknowledging inflight message. ClientId={}, messageId={}", clientID, messageID); - Map m = this.m_inflightStore.get(clientID); + public StoredMessage inFlightAck(String clientID, int messageID) { + LOG.debug("Acknowledging inflight message CId={}, messageId={}", clientID, messageID); + ConcurrentMap m = this.outboundFlightMessages.get(clientID); if (m == null) { - LOG.warn("Unable to retrieve inflight message record. ClientId={}, messageId={}", clientID, messageID); - return; + LOG.error("Can't find the inFlight record for client <{}>", clientID); + throw new RuntimeException("Can't find the inFlight record for client <" + clientID + ">"); } - m.remove(messageID); + StoredMessage msg = m.remove(messageID); + this.outboundFlightMessages.put(clientID, m); // remove from the ids store Set inFlightForClient = this.m_inFlightIds.get(clientID); if (inFlightForClient != null) { inFlightForClient.remove(messageID); } + return msg; } @Override - public void inFlight(String clientID, int messageID, MessageGUID guid) { - LOG.debug("Storing inflight message. ClientId={}, messageId={}, guid={}", clientID, messageID, guid); - ConcurrentMap m = this.m_inflightStore.get(clientID); - if (m == null) { - m = new ConcurrentHashMap<>(); + public void inFlight(String clientID, int messageID, StoredMessage msg) { + ConcurrentMap messages = outboundFlightMessages.get(clientID); + if (messages == null) { + messages = new ConcurrentHashMap<>(); } - m.put(messageID, guid); - LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid); - this.m_inflightStore.put(clientID, m); + messages.put(messageID, msg); + outboundFlightMessages.put(clientID, messages); } @Override @@ -250,78 +247,67 @@ class MapDBSessionsStore implements ISessionsStore { } @Override - public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Moving inflight message to 2nd phase ack state. ClientId={}, messageID={}", clientID, messageID); - } - Map m = this.m_inflightStore.get(clientID); + public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) { + LOG.debug("Moving inflight message to 2nd phase ack state. ClientId={}, messageID={}", clientID, messageID); + ConcurrentMap m = this.m_secondPhaseStore.get(clientID); if (m == null) { - LOG.warn("Unable to retrieve inflight message record. ClientId={}, messageId={}", clientID, messageID); - return; + String error = String.format("Can't find the inFlight record for client <%s> during the second phase of " + + "QoS2 pub", clientID); + LOG.error(error); + throw new RuntimeException(error); } - MessageGUID guid = m.remove(messageID); - - // remove from the ids store - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient != null) { - inFlightForClient.remove(messageID); - } - - final HashMap emptyGuids = new HashMap<>(); - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); - messageIDs.put(messageID, guid); - m_secondPhaseStore.put(clientID, messageIDs); + m.put(messageID, msg); + this.outboundFlightMessages.put(clientID, m); } @Override - public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); + public StoredMessage secondPhaseAcknowledged(String clientID, int messageID) { + LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); + final ConcurrentMap m = this.m_secondPhaseStore.get(clientID); + if (m == null) { + String error = String.format("Can't find the inFlight record for client <%s> during the second phase " + + "acking of QoS2 pub", clientID); + LOG.error(error); + throw new RuntimeException(error); } - final HashMap emptyGuids = new HashMap<>(); - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, emptyGuids); - MessageGUID guid = messageIDs.remove(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - return guid; - } - @Override - public StoredMessage getInflightMessage(String clientID, int messageID) { - LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID); - Map clientEntries = m_inflightStore.get(clientID); - if (clientEntries == null) { - LOG.warn("The client has no inflight messages CId={}, messageId={}", clientID, messageID); - return null; - } - MessageGUID guid = clientEntries.get(messageID); - if (guid == null) { - LOG.warn("The message ID does not have an associated GUID. ClientId={}, messageId={}", clientID, messageID); - return null; - } - return m_messagesStore.getMessageByGuid(guid); + StoredMessage msg = m.remove(messageID); + m_secondPhaseStore.put(clientID, m); + return msg; } @Override public int getInflightMessagesNo(String clientID) { - if (!m_inflightStore.containsKey(clientID)) - return 0; - else - return m_inflightStore.get(clientID).size(); + int totalInflight = 0; + ConcurrentMap inflightPerClient = m_db.getHashMap(inboundMessageId2MessagesMapName(clientID)); + if (inflightPerClient != null) { + totalInflight += inflightPerClient.size(); + } + + Map secondPhaseInFlight = this.m_secondPhaseStore.get(clientID); + if (secondPhaseInFlight != null) { + totalInflight += secondPhaseInFlight.size(); + } + + Map outboundPerClient = outboundFlightMessages.get(clientID); + if (outboundPerClient != null) { + totalInflight += outboundPerClient.size(); + } + + return totalInflight; } @Override public StoredMessage inboundInflight(String clientID, int messageID) { LOG.debug("Mapping inbound message ID to GUID CId={}, messageId={}", clientID, messageID); - ConcurrentMap messageIdToGuid = m_db.getHashMap(inboundMessageId2GuidsMapName(clientID)); - final MessageGUID guid = messageIdToGuid.get(messageID); - LOG.debug("Inbound message ID has been mapped CId={}, messageId={}, guid={}", clientID, messageID, guid); - return m_messagesStore.getMessageByGuid(guid); + ConcurrentMap messageIdToGuid = m_db.getHashMap(inboundMessageId2MessagesMapName(clientID)); + return messageIdToGuid.get(messageID); } @Override - public void markAsInboundInflight(String clientID, int messageID, MessageGUID guid) { - ConcurrentMap messageIdToGuid = m_db.getHashMap(inboundMessageId2GuidsMapName(clientID)); - messageIdToGuid.put(messageID, guid); + public void markAsInboundInflight(String clientID, int messageID, StoredMessage msg) { + ConcurrentMap messageIdToGuid = m_db.getHashMap(inboundMessageId2MessagesMapName(clientID)); + messageIdToGuid.put(messageID, msg); } @Override @@ -333,25 +319,16 @@ class MapDBSessionsStore implements ISessionsStore { public int getSecondPhaseAckPendingMessages(String clientID) { if (!m_secondPhaseStore.containsKey(clientID)) return 0; - else - return m_secondPhaseStore.get(clientID).size(); + return m_secondPhaseStore.get(clientID).size(); } @Override - public Collection pendingAck(String clientID) { - ConcurrentMap messageGUIDMap = m_db.getHashMap(messageId2GuidsMapName(clientID)); - if (messageGUIDMap == null || messageGUIDMap.isEmpty()) { - return Collections.emptyList(); - } - - return new ArrayList<>(messageGUIDMap.values()); + public void dropInFlightMessagesInSession(String clientID) { + m_secondPhaseStore.remove(clientID); + outboundFlightMessages.remove(clientID); } - static String messageId2GuidsMapName(String clientID) { - return "guidsMapping_" + clientID; - } - - static String inboundMessageId2GuidsMapName(String clientID) { - return "inboundInflightGuidsMapping_" + clientID; + static String inboundMessageId2MessagesMapName(String clientID) { + return "inboundInflight_" + clientID; } } diff --git a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java index 11b6338a..958d26b5 100644 --- a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java +++ b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java @@ -17,40 +17,30 @@ package io.moquette.persistence.mapdb; import io.moquette.BrokerConstants; +import io.moquette.persistence.MessageStoreTCK; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; -import io.moquette.spi.ClientSession; -import io.moquette.spi.IMessagesStore; -import io.moquette.spi.ISessionsStore; -import io.moquette.spi.ISessionsStore.ClientTopicCouple; -import io.moquette.spi.MessageGUID; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.Topic; -import static io.netty.handler.codec.mqtt.MqttQoS.*; +import io.moquette.spi.IMessagesStore.StoredMessage; +import io.netty.handler.codec.mqtt.MqttQoS; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.io.File; -import java.util.List; import java.util.Properties; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import static java.util.Collections.singletonList; import static org.junit.Assert.*; /** * * @author andrea */ -public class MapDBPersistentStoreTest { +public class MapDBPersistentStoreTest extends MessageStoreTCK { - public static final String TEST_CLIENT = "TestClient"; MapDBPersistentStore m_storageService; - ISessionsStore m_sessionsStore; - IMessagesStore m_messagesStore; private ScheduledExecutorService scheduler; @@ -64,8 +54,8 @@ public class MapDBPersistentStoreTest { IConfig conf = new MemoryConfig(props); m_storageService = new MapDBPersistentStore(conf, scheduler); m_storageService.initStore(); - m_messagesStore = m_storageService.messagesStore(); - m_sessionsStore = m_storageService.sessionsStore(); + messagesStore = m_storageService.messagesStore(); + sessionsStore = m_storageService.sessionsStore(); } @After @@ -88,56 +78,38 @@ public class MapDBPersistentStoreTest { assertFalse(dbFile.exists()); } - @Test - public void overridingSubscriptions() { - ClientSession session1 = m_sessionsStore.createNewSession("SESSION_ID_1", true); - - // Subscribe on /topic with QOSType.MOST_ONE - Subscription oldSubscription = new Subscription(session1.clientID, new Topic("/topic"), AT_MOST_ONCE); - session1.subscribe(oldSubscription); - - // Subscribe on /topic again that overrides the previous subscription. - Subscription overridingSubscription = new Subscription(session1.clientID, new Topic("/topic"), EXACTLY_ONCE); - session1.subscribe(overridingSubscription); - - // Verify - List subscriptions = m_sessionsStore.listAllSubscriptions(); - assertEquals(1, subscriptions.size()); - Subscription sub = m_sessionsStore.getSubscription(subscriptions.get(0)); - assertEquals(overridingSubscription.getRequestedQos(), sub.getRequestedQos()); - } - @Test public void testNextPacketID_notExistingClientSession() { - int packetId = m_sessionsStore.nextPacketID("NOT_EXISTING_CLI"); + int packetId = sessionsStore.nextPacketID("NOT_EXISTING_CLI"); assertEquals(1, packetId); } @Test public void testNextPacketID_existingClientSession() { // Force creation of inflight map for the CLIENT session - int packetId = m_sessionsStore.nextPacketID("CLIENT"); + int packetId = sessionsStore.nextPacketID("CLIENT"); assertEquals(1, packetId); // request a second packetID - packetId = m_sessionsStore.nextPacketID("CLIENT"); + packetId = sessionsStore.nextPacketID("CLIENT"); assertEquals(2, packetId); } @Test public void testNextPacketID() { - // request a first ID + StoredMessage msgStored = new StoredMessage("Hello".getBytes(), MqttQoS.AT_LEAST_ONCE, "/topic"); + msgStored.setClientID(TEST_CLIENT); - int packetId = m_sessionsStore.nextPacketID("CLIENT"); - m_sessionsStore.inFlight("CLIENT", packetId, new MessageGUID("ABCDE")); // simulate an - // inflight + // request a first ID + int packetId = sessionsStore.nextPacketID("CLIENT"); + sessionsStore.inFlight("CLIENT", packetId, msgStored); // simulate an inflight assertEquals(1, packetId); // release the ID - m_sessionsStore.inFlightAck("CLIENT", packetId); + sessionsStore.inFlightAck("CLIENT", packetId); // request a second packetID, counter restarts from 0 - packetId = m_sessionsStore.nextPacketID("CLIENT"); + packetId = sessionsStore.nextPacketID("CLIENT"); assertEquals(1, packetId); } @@ -146,44 +118,8 @@ public class MapDBPersistentStoreTest { m_storageService.close(); // verify the executor is shutdown - assertTrue( - "Storage service scheduler can't be stopped in 3 seconds", + assertTrue("Storage service scheduler can't be stopped in 3 seconds", m_storageService.m_scheduler.awaitTermination(3, TimeUnit.SECONDS)); assertTrue(m_storageService.m_scheduler.isTerminated()); } - - @Test - public void testDropMessagesInSessionCleanAllNotRetainedStoredMessages() { - m_sessionsStore.createNewSession(TEST_CLIENT, true); - IMessagesStore.StoredMessage publishToStore = new IMessagesStore.StoredMessage("Hello".getBytes(), EXACTLY_ONCE, - "/topic"); - publishToStore.setClientID(TEST_CLIENT); - publishToStore.setRetained(false); - MessageGUID guid = m_messagesStore.storePublishForFuture(publishToStore); - - // Exercise - m_messagesStore.dropInFlightMessagesInSession(singletonList(guid)); - - // Verify the message store for session is empty. - IMessagesStore.StoredMessage storedPublish = m_messagesStore.getMessageByGuid(guid); - assertNull("The stored message must'n be present anymore", storedPublish); - } - - @Test - public void testDropMessagesInSessionDoesntCleanAnyRetainedStoredMessages() { - m_sessionsStore.createNewSession(TEST_CLIENT, true); - IMessagesStore.StoredMessage publishToStore = new IMessagesStore.StoredMessage("Hello".getBytes(), EXACTLY_ONCE, - "/topic"); - publishToStore.setClientID(TEST_CLIENT); - publishToStore.setRetained(true); - MessageGUID guid = m_messagesStore.storePublishForFuture(publishToStore); - m_messagesStore.storeRetained(new Topic("/topic"), guid); - - // Exercise - m_messagesStore.dropInFlightMessagesInSession(singletonList(guid)); - - // Verify the message store for session is empty. - IMessagesStore.StoredMessage storedPublish = m_messagesStore.getMessageByGuid(guid); - assertNotNull("The stored retained message must be present after client's session drop", storedPublish); - } } diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java index 08e58b7d..ce7d18db 100644 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java @@ -34,7 +34,7 @@ public final class PerformanceTest { cpufreq-set -g performance -u 800MHz -c 3 */ public static void main(String[] args) throws IOException, InterruptedException { - SubscriptionsStore store = new SubscriptionsStore(); + SubscriptionsDirectory store = new SubscriptionsDirectory(); MemoryStorageService memStore = new MemoryStorageService(); ISessionsStore aSessionsStore = memStore.sessionsStore(); store.init(aSessionsStore); @@ -56,7 +56,7 @@ public final class PerformanceTest { + ((i / (users * devices * values)) % subs)); Subscription sub = new Subscription("CLI_ID_" + (i % users), topics[i], MqttQoS.AT_MOST_ONCE); - aSessionsStore.addNewSubscription(sub); + aSessionsStore.subscriptionStore().addNewSubscription(sub); store.add(sub.asClientTopicCouple()); } diff --git a/pom.xml b/pom.xml index 6b5b9c91..548df226 100644 --- a/pom.xml +++ b/pom.xml @@ -110,6 +110,7 @@ broker mapdb_storage + h2_storage distribution embedding_moquette osgi_test