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