From 9ce9a6597fa5d92fbfde422b8ae57e796ac56237 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 6 Dec 2015 11:43:29 +0100 Subject: [PATCH] Architechtural refactory, moved subscription out of tree and into db store --- .../server/netty/NettyMQTTHandler.java | 2 +- .../java/io/moquette/spi/ClientSession.java | 12 +- .../java/io/moquette/spi/ISessionsStore.java | 47 ++++++- .../moquette/spi/impl/ProtocolProcessor.java | 13 +- .../spi/impl/subscriptions/Subscription.java | 13 +- .../subscriptions/SubscriptionsStore.java | 43 +++---- .../spi/impl/subscriptions/TreeNode.java | 62 +++------- .../spi/persistence/MapDBSessionsStore.java | 86 +++++-------- .../server/ServerIntegrationPahoTest.java | 8 +- .../spi/impl/ProtocolProcessorTest.java | 9 +- .../subscriptions/SubscriptionsStoreTest.java | 117 ++++++++++++------ .../spi/impl/subscriptions/TreeNodeTest.java | 33 ----- .../persistence/MapDBPersistentStoreTest.java | 11 +- .../spi/persistence/MemorySessionStore.java | 34 +++-- broker/src/test/resources/log4j.properties | 20 +-- 15 files changed, 260 insertions(+), 250 deletions(-) delete mode 100644 broker/src/test/java/io/moquette/spi/impl/subscriptions/TreeNodeTest.java diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index f76a924e..a10c53c1 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -98,7 +98,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { } m_processor.processConnectionLost(clientID, stolen); } - ctx.close(/*false*/); + ctx.close(); } @Override diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 4cba7048..0a3ccab9 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -99,8 +99,16 @@ public class ClientSession { //send SUBACK with 0x80 for this topic filter return false; } - subscriptions.add(newSubscription); - m_sessionsStore.addNewSubscription(newSubscription); + ISessionsStore.ClientTopicCouple matchingCouple = new ISessionsStore.ClientTopicCouple(this.clientID, newSubscription.getTopicFilter()); + Subscription existingSub = m_sessionsStore.getSubscription(matchingCouple); + //update the selected subscriptions if not present or if has a greater qos + if (existingSub == null || existingSub.getRequestedQos().byteValue() < newSubscription.getRequestedQos().byteValue()) { + if (existingSub != null) { + subscriptions.remove(newSubscription); + } + subscriptions.add(newSubscription); + m_sessionsStore.addNewSubscription(newSubscription); + } return true; } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index ec2068a6..9308e0e1 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -17,7 +17,6 @@ package io.moquette.spi; import java.util.Collection; import java.util.List; -import java.util.Set; import io.moquette.spi.impl.subscriptions.Subscription; @@ -28,6 +27,43 @@ import io.moquette.spi.impl.subscriptions.Subscription; */ public interface ISessionsStore { + class ClientTopicCouple { + public final String topicFilter; + public final String clientID; + + public ClientTopicCouple(String clientID, String 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(); /** @@ -46,11 +82,14 @@ public interface ISessionsStore { void wipeSubscriptions(String sessionID); /** - * Updates the subscriptions set for the clientID + * Return all topic filters to recreate the subscription tree. * */ - void updateSubscriptions(String clientID, Set subscriptions); + List listAllSubscriptions(); - List listAllSubscriptions(); + /** + * @return the subscription stored by clientID and topicFilter, if any else null; + * */ + Subscription getSubscription(ClientTopicCouple couple); /** * @return true iff there are subscriptions persisted with clientID 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 b47c51f3..9d85b3a6 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -26,6 +26,7 @@ import io.moquette.spi.ClientSession; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISessionsStore.ClientTopicCouple; import io.moquette.spi.impl.security.IAuthenticator; import io.moquette.spi.impl.security.IAuthorizator; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -383,6 +384,7 @@ public class ProtocolProcessor { } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it + //TODO isCleanSession MUST be on the targetSession not the Subscription if (!sub.isCleanSession() && !targetSession.isActive()) { //store the message in targetSession queue to deliver targetSession.enqueueToDeliver(guid); @@ -621,18 +623,19 @@ public class ProtocolProcessor { //fire the publish for(Subscription subscription : newSubscriptions) { - subscribeSingleTopic(subscription, subscription.getTopicFilter()); + subscribeSingleTopic(subscription); } } - private boolean subscribeSingleTopic(Subscription newSubscription, final String topic) { - subscriptions.add(newSubscription); + private boolean subscribeSingleTopic(final Subscription newSubscription) { + final ClientTopicCouple couple = new ClientTopicCouple(newSubscription.getClientId(), newSubscription.getTopicFilter()); + subscriptions.add(couple); //scans retained messages to be published to the new subscription //TODO this is ugly, it does a linear scan on potential big dataset Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { public boolean match(String key) { - return SubscriptionsStore.matchTopics(key, topic); + return SubscriptionsStore.matchTopics(key, newSubscription.getTopicFilter()); } }); @@ -640,7 +643,7 @@ public class ProtocolProcessor { for (IMessagesStore.StoredMessage storedMsg : messages) { //fire the as retained the message - LOG.debug("send publish message for topic {}", topic); + LOG.debug("send publish message for topic {}", newSubscription.getTopicFilter()); //forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null : m_messagesStore.nextPacketID(newSubscription.getClientId()); 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 fb10d173..01ca9008 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 @@ -60,6 +60,10 @@ public class Subscription implements Serializable { return topicFilter; } + /** + * @deprecated this concept must be moved inside ClientSession and not subscription. + * */ + @Deprecated public boolean isCleanSession() { return this.cleanSession; } @@ -71,20 +75,15 @@ public class Subscription implements Serializable { Subscription that = (Subscription) o; - if (cleanSession != that.cleanSession) return false; if (clientId != null ? !clientId.equals(that.clientId) : that.clientId != null) return false; - if (requestedQos != that.requestedQos) return false; - if (topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null) return false; + return !(topicFilter != null ? !topicFilter.equals(that.topicFilter) : that.topicFilter != null); - return true; } @Override public int hashCode() { - int result = requestedQos.hashCode(); - result = 31 * result + (clientId != null ? clientId.hashCode() : 0); + int result = clientId != null ? clientId.hashCode() : 0; result = 31 * result + (topicFilter != null ? topicFilter.hashCode() : 0); - result = 31 * result + (cleanSession ? 1 : 0); return result; } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 914722bc..2fce9744 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -21,6 +21,7 @@ import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.atomic.AtomicReference; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISessionsStore.ClientTopicCouple; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -68,8 +69,8 @@ public class SubscriptionsStore { public void visit(TreeNode node, int deep) { String subScriptionsStr = ""; String indentTabs = indentTabs(deep); - for (Subscription sub : node.m_subscriptions) { - subScriptionsStr += indentTabs + sub.toString() + "\n"; + for (ClientTopicCouple couple : node.m_subscriptions) { + subScriptionsStr += indentTabs + couple.toString() + "\n"; } s += node.getToken() == null ? "" : node.getToken().toString(); s += "\n" + (node.m_subscriptions.isEmpty() ? indentTabs : "") + subScriptionsStr /*+ "\n"*/; @@ -92,6 +93,7 @@ public class SubscriptionsStore { private AtomicReference subscriptions = new AtomicReference<>(new TreeNode()); private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); + private volatile ISessionsStore m_sessionsStore; /** * Initialize the subscription tree with the list of subscriptions. @@ -99,22 +101,23 @@ public class SubscriptionsStore { */ public void init(ISessionsStore sessionsStore) { LOG.debug("init invoked"); - List subscriptions = sessionsStore.listAllSubscriptions(); + m_sessionsStore = sessionsStore; + List subscriptions = sessionsStore.listAllSubscriptions(); //reload any subscriptions persisted if (LOG.isDebugEnabled()) { LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree()); } - for (Subscription subscription : subscriptions) { - LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter()); - add(subscription); + for (ClientTopicCouple clientTopic : subscriptions) { + LOG.debug("Re-subscribing {} to topic {}", clientTopic.clientID, clientTopic.topicFilter); + add(clientTopic); } if (LOG.isTraceEnabled()) { LOG.trace("Finished loading. Subscription tree after {}", dumpTree()); } } - public void add(Subscription newSubscription) { + public void add(ClientTopicCouple newSubscription) { TreeNode oldRoot; NodeCouple couple; do { @@ -146,13 +149,12 @@ public class SubscriptionsStore { if ((matchingChildren = current.childWithToken(token)) != null) { //copy the traversed node current = matchingChildren.copy(); -// current.m_parent = parent; //update the child just added in the children list parent.updateChild(matchingChildren, current); parent = current; } else { //create a new node for the newly inserted token - matchingChildren = new TreeNode(/*current*/); + matchingChildren = new TreeNode(); matchingChildren.setToken(token); current.addChild(matchingChildren); current = matchingChildren; @@ -168,19 +170,7 @@ public class SubscriptionsStore { oldRoot = subscriptions.get(); couple = recreatePath(topic, oldRoot); - //do the job - //search for the subscription to remove - Subscription toBeRemoved = null; - for (Subscription sub : couple.createdNode.subscriptions()) { - if (sub.topicFilter.equals(topic) && sub.getClientId().equals(clientID)) { - toBeRemoved = sub; - break; - } - } - - if (toBeRemoved != null) { - couple.createdNode.subscriptions().remove(toBeRemoved); - } + couple.createdNode.remove(new ClientTopicCouple(clientID, topic)); //spin lock repeating till we can, swap root, if can't swap just re-do the operation } while(!subscriptions.compareAndSet(oldRoot, couple.root)); } @@ -216,16 +206,17 @@ public class SubscriptionsStore { } Queue tokenQueue = new LinkedBlockingDeque<>(tokens); - List matchingSubs = new ArrayList<>(); + List matchingSubs = new ArrayList<>(); subscriptions.get().matches(tokenQueue, matchingSubs); //remove the overlapping subscriptions, selecting ones with greatest qos Map subsForClient = new HashMap<>(); - for (Subscription sub : matchingSubs) { - Subscription existingSub = subsForClient.get(sub.getClientId()); + for (ClientTopicCouple matchingCouple : matchingSubs) { + Subscription existingSub = subsForClient.get(matchingCouple.clientID); + Subscription sub = m_sessionsStore.getSubscription(matchingCouple); //update the selected subscriptions if not present or if has a greater qos if (existingSub == null || existingSub.getRequestedQos().byteValue() < sub.getRequestedQos().byteValue()) { - subsForClient.put(sub.getClientId(), sub); + subsForClient.put(matchingCouple.clientID, sub); } } return new ArrayList<>(subsForClient.values()); 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 657d7b02..a7550020 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 @@ -15,22 +15,17 @@ */ package io.moquette.spi.impl.subscriptions; +import io.moquette.spi.ISessionsStore.ClientTopicCouple; + import java.util.*; import java.util.concurrent.LinkedBlockingQueue; class TreeNode { - private class ClientIDComparator implements Comparator { - - public int compare(Subscription o1, Subscription o2) { - return o1.getClientId().compareTo(o2.getClientId()); - } - - } - Token m_token; List m_children = new ArrayList<>(); - List m_subscriptions = new ArrayList<>(); + //TODO move to set of ClientIDthe set of clientIDs that has subscriptions to this topic + Set m_subscriptions = new HashSet<>(); TreeNode() { } @@ -43,19 +38,7 @@ class TreeNode { this.m_token = topic; } - void addSubscription(Subscription s) { - //avoid double registering for same clientID, topic and QoS - if (m_subscriptions.contains(s)) { - return; - } - //remove existing subscription for same client and topic but different QoS - Comparator comparator = new ClientIDComparator(); - Collections.sort(m_subscriptions, comparator); - int existingSubIdx = Collections.binarySearch(m_subscriptions, s, comparator); - if (existingSubIdx >= 0) { - m_subscriptions.remove(existingSubIdx); - } - + void addSubscription(ClientTopicCouple s) { m_subscriptions.add(s); } @@ -70,7 +53,7 @@ class TreeNode { TreeNode copy() { final TreeNode copy = new TreeNode(); copy.m_children = new ArrayList<>(m_children); - copy.m_subscriptions = new ArrayList<>(m_subscriptions); + copy.m_subscriptions = new HashSet<>(m_subscriptions); copy.m_token = m_token; return copy; } @@ -94,11 +77,15 @@ class TreeNode { m_children.add(newChild); } - List subscriptions() { + Collection subscriptions() { return m_subscriptions; } - void matches(Queue tokens, List matchingSubs) { + public void remove(ClientTopicCouple clientTopicCouple) { + m_subscriptions.remove(clientTopicCouple); + } + + void matches(Queue tokens, List matchingSubs) { Token t = tokens.poll(); //check if t is null <=> tokens finished @@ -147,14 +134,14 @@ class TreeNode { TreeNode removeClientSubscriptions(String clientID) { //collect what to delete and then delete to avoid ConcurrentModification TreeNode newSubRoot = this.copy(); - List subsToRemove = new ArrayList<>(); - for (Subscription s : newSubRoot.m_subscriptions) { - if (s.clientId.equals(clientID)) { + List subsToRemove = new ArrayList<>(); + for (ClientTopicCouple s : newSubRoot.m_subscriptions) { + if (s.clientID.equals(clientID)) { subsToRemove.add(s); } } - for (Subscription s : subsToRemove) { + for (ClientTopicCouple s : subsToRemove) { newSubRoot.m_subscriptions.remove(s); } @@ -166,21 +153,4 @@ class TreeNode { newSubRoot.m_children = newChildren; return newSubRoot; } - - /** - * @return the set of subscriptions for the given client. - * */ - Set findAllByClientID(String clientID) { - Set subs = new HashSet<>(); - for (Subscription s : m_subscriptions) { - if (s.clientId.equals(clientID)) { - subs.add(s); - } - } - //go deep - for (TreeNode child : m_children) { - subs.addAll(child.findAllByClientID(clientID)); - } - return subs; - } } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index c20c02d1..728a1f27 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -41,8 +41,6 @@ class MapDBSessionsStore implements ISessionsStore { private ConcurrentMap> m_inflightStore; //map clientID <-> set of currently in flight packet identifiers private Map> m_inFlightIds; - //persistent Map of clientID, set of Subscriptions - private ConcurrentMap> m_persistentSubscriptions; private ConcurrentMap m_persistentSessions; //maps clientID->[guid*], insertion order cares, it's queue private ConcurrentMap> m_enqueuedStore; @@ -61,7 +59,6 @@ class MapDBSessionsStore implements ISessionsStore { public void initStore() { m_inflightStore = m_db.getHashMap("inflight"); m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); - m_persistentSubscriptions = m_db.getHashMap("subscriptions"); m_persistentSessions = m_db.getHashMap("sessions"); m_enqueuedStore = m_db.getHashMap("sessionQueue"); m_secondPhaseStore = m_db.getHashMap("secondPhase"); @@ -71,79 +68,57 @@ class MapDBSessionsStore implements ISessionsStore { public void addNewSubscription(Subscription newSubscription) { LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); final String clientID = newSubscription.getClientId(); - if (!m_persistentSubscriptions.containsKey(clientID)) { - LOG.debug("subscriptions for clientID <{}> not present, empty subscriptions set", clientID); - m_persistentSubscriptions.put(clientID, new HashSet()); - } + m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); - Set subs = m_persistentSubscriptions.get(clientID); - if (!subs.contains(newSubscription)) { - LOG.debug("updating <{}> subscriptions set with new subscription", clientID); - //TODO check the subs doesn't contain another subscription to the same topic with different - Subscription existingSubscription = null; - for (Subscription scanSub : subs) { - if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) { - existingSubscription = scanSub; - break; - } - } - if (existingSubscription != null) { - subs.remove(existingSubscription); - } - subs.add(newSubscription); - m_persistentSubscriptions.put(clientID, subs); - LOG.debug("clientID {} subscriptions set now is {}", clientID, subs); + if (LOG.isTraceEnabled()) { + LOG.trace("subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); } } @Override - public void removeSubscription(String topic, String clientID) { - LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); - if (!m_persistentSubscriptions.containsKey(clientID)) { + public void removeSubscription(String topicFilter, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topicFilter, clientID); + if (!m_db.exists("subscriptions_" + clientID)) { return; } - Set clientSubscriptions = m_persistentSubscriptions.get(clientID); - //search for the subscription to remove - Subscription toBeRemoved = null; - for (Subscription sub : clientSubscriptions) { - if (sub.getTopicFilter().equals(topic)) { - toBeRemoved = sub; - break; - } - } - - if (toBeRemoved != null) { - clientSubscriptions.remove(toBeRemoved); - } - m_persistentSubscriptions.put(clientID, clientSubscriptions); - if (LOG.isTraceEnabled()) { - LOG.trace("persisted subscriptions {}", m_persistentSubscriptions); - } + m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); } @Override public void wipeSubscriptions(String clientID) { - m_persistentSubscriptions.remove(clientID); + LOG.debug("wipeSubscriptions"); + if (LOG.isTraceEnabled()) { + LOG.trace("Subscription pre wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); + } + m_db.delete("subscriptions_" + clientID); + if (LOG.isTraceEnabled()) { + LOG.trace("Subscription post wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); + } } @Override - public void updateSubscriptions(String clientID, Set subscriptions) { - m_persistentSubscriptions.put(clientID, subscriptions); - } - - @Override - public List listAllSubscriptions() { - List allSubscriptions = new ArrayList<>(); - for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { - allSubscriptions.addAll(entry.getValue()); + public List listAllSubscriptions() { + final List allSubscriptions = new ArrayList<>(); + for (String clientID : m_persistentSessions.keySet()) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); + for (String topicFilter : clientSubscriptions.keySet()) { + allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); + } } LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions); return allSubscriptions; } + @Override + public Subscription getSubscription(ClientTopicCouple couple) { + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); + LOG.debug("subscriptions_{}: {}", couple.clientID, clientSubscriptions); + return clientSubscriptions.get(couple.topicFilter); + } + @Override public boolean contains(String clientID) { - return m_persistentSubscriptions.containsKey(clientID); + return m_db.exists("subscriptions_" + clientID); } @Override @@ -154,7 +129,6 @@ class MapDBSessionsStore implements ISessionsStore { throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); } LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); - m_persistentSubscriptions.put(clientID, new HashSet()); m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false)); return new ClientSession(clientID, m_messagesStore, this, cleanSession); } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 00bc27d8..5ab9acad 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -62,9 +62,7 @@ public class ServerIntegrationPahoTest { public void setUp() throws Exception { String dbPath = IntegrationUtils.localMapDBPath(); IntegrationUtils.cleanPersistenceFile(dbPath); -// File dbFile = new File(dbPath); -// assertFalse(String.format("The DB storagefile %s already exists", dbPath), dbFile.exists()); - + startServer(); m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); @@ -139,7 +137,7 @@ public class ServerIntegrationPahoTest { /** * Check that after a client has connected with clean session false, subscribed - * to some topic and exited, if it reconnect with clean session true, the m_server + * to some topic and exited, if it reconnects with clean session true, the m_server * correctly cleanup every previous subscription */ @Test @@ -152,7 +150,7 @@ public class ServerIntegrationPahoTest { m_client.disconnect(); //reconnect and publish - m_client.connect(); + m_client.connect(options); m_client.publish("/topic", "Test my payload".getBytes(), 0, false); assertEquals("Test my payload", new String(m_callback.getMessage(false).getPayload())); 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 29715350..83bffc4c 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -24,6 +24,7 @@ import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISessionsStore.ClientTopicCouple; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -340,7 +341,7 @@ public class ProtocolProcessorTest { //Exercise SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple(QOSType.MOST_ONE.byteValue(), "#")); - m_processor.processSubscribe(m_session, msg/*, FAKE_PUBLISHER_ID, false*/); + m_processor.processSubscribe(m_session, msg); //Verify //wait the latch @@ -483,10 +484,12 @@ public class ProtocolProcessorTest { Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false); Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false); + sessionsStore.addNewSubscription(subQos1); + sessionsStore.addNewSubscription(subQos2); SubscriptionsStore subscriptions = new SubscriptionsStore(); subscriptions.init(sessionsStore); - subscriptions.add(subQos1); - subscriptions.add(subQos2); + subscriptions.add(new ClientTopicCouple("Sub A", "a/b")); + subscriptions.add(new ClientTopicCouple("Sub B", "a/+")); ProtocolProcessor processor = new ProtocolProcessor() { diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index a9f9518c..d19b968b 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -20,7 +20,9 @@ import java.text.ParseException; import java.util.Arrays; import java.util.List; +import io.moquette.spi.ClientSession; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISessionsStore.ClientTopicCouple; import io.moquette.spi.impl.MemoryStorageService; import io.moquette.proto.messages.AbstractMessage; @@ -35,6 +37,7 @@ import org.junit.Test; public class SubscriptionsStoreTest { private SubscriptionsStore store; + private ISessionsStore sessionsStore; public SubscriptionsStoreTest() { } @@ -44,7 +47,8 @@ public class SubscriptionsStoreTest { store = new SubscriptionsStore(); MemoryStorageService storageService = new MemoryStorageService(); storageService.initStore(); - store.init(storageService.sessionsStore()); + this.sessionsStore = storageService.sessionsStore(); + store.init(sessionsStore); } @Test @@ -103,11 +107,13 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimple() { Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE, false); - store.add(slashSub); + sessionsStore.addNewSubscription(slashSub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "/")); assertTrue(store.matches("finance").isEmpty()); - + Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE, false); - store.add(slashFinanceSub); + sessionsStore.addNewSubscription(slashFinanceSub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "/finance")); assertTrue(store.matches("finance").isEmpty()); assertTrue(store.matches("/finance").contains(slashFinanceSub)); @@ -117,11 +123,13 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti() { Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); - store.add(anySub); + sessionsStore.addNewSubscription(anySub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "#")); assertTrue(store.matches("finance").contains(anySub)); Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); - store.add(financeAnySub); + sessionsStore.addNewSubscription(financeAnySub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_2", "finance/#")); assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub))); } @@ -129,8 +137,10 @@ public class SubscriptionsStoreTest { public void testMatchingDeepMulti_one_layer() { Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); - store.add(anySub); - store.add(financeAnySub); + sessionsStore.addNewSubscription(anySub); + sessionsStore.addNewSubscription(financeAnySub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "#")); + store.add(new ClientTopicCouple("FAKE_CLI_ID_2", "finance/#")); //Verify assertTrue(store.matches("finance/stock").containsAll(Arrays.asList(financeAnySub, anySub))); @@ -141,7 +151,8 @@ public class SubscriptionsStoreTest { @Test public void testMatchingDeepMulti_two_layer() { Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE, false); - store.add(financeAnySub); + sessionsStore.addNewSubscription(financeAnySub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "finance/stock/#")); //Verify assertTrue(store.matches("finance/stock/ibm").contains(financeAnySub)); @@ -150,18 +161,21 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleSingle() { Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false); - store.add(anySub); + sessionsStore.addNewSubscription(anySub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "+")); assertTrue(store.matches("finance").contains(anySub)); Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE, false); - store.add(financeOne); + sessionsStore.addNewSubscription(financeOne); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "finance/+")); assertTrue(store.matches("finance/stock").contains(financeOne)); } @Test public void testMatchManySingle() { Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE, false); - store.add(manySub); + sessionsStore.addNewSubscription(manySub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "+/+")); //verify assertTrue(store.matches("/finance").contains(manySub)); @@ -171,9 +185,11 @@ public class SubscriptionsStoreTest { @Test public void testMatchSlashSingle() { Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE, false); - store.add(slashPlusSub); + sessionsStore.addNewSubscription(slashPlusSub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "/+")); Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false); - store.add(anySub); + sessionsStore.addNewSubscription(anySub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "+")); //Verify assertEquals(1, store.matches("/finance").size()); @@ -185,10 +201,12 @@ public class SubscriptionsStoreTest { @Test public void testMatchManyDeepSingle() { Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE, false); - store.add(slashPlusSub); + sessionsStore.addNewSubscription(slashPlusSub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "/finance/+/ibm")); Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_2", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE, false); - store.add(slashPlusDeepSub); + sessionsStore.addNewSubscription(slashPlusDeepSub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_2", "/+/stock/+")); //Verify assertTrue(store.matches("/finance/stock/ibm").containsAll(Arrays.asList(slashPlusSub, slashPlusDeepSub))); @@ -196,7 +214,9 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_allTheTree() { - store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false)); + Subscription sub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); + sessionsStore.addNewSubscription(sub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "#")); assertFalse(store.matches("finance").isEmpty()); assertFalse(store.matches("finance/ibm").isEmpty()); } @@ -204,7 +224,9 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_zeroLevel() { //check MULTI in case of zero level match - store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); + Subscription sub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); + sessionsStore.addNewSubscription(sub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "finance/#")); assertFalse(store.matches("finance").isEmpty()); } @@ -236,9 +258,11 @@ public class SubscriptionsStoreTest { store = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); memStore.initStore(); - store.init(memStore.sessionsStore()); + ISessionsStore aSessionsStore = memStore.sessionsStore(); + store.init(aSessionsStore); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); - store.add(sub); + aSessionsStore.addNewSubscription(sub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", subscription)); assertFalse(store.matches(topic).isEmpty()); } @@ -248,7 +272,8 @@ public class SubscriptionsStoreTest { memStore.initStore(); store.init(memStore.sessionsStore()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); - store.add(sub); + sessionsStore.addNewSubscription(sub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", subscription)); assertTrue(store.matches(topic).isEmpty()); } @@ -256,7 +281,8 @@ public class SubscriptionsStoreTest { @Test public void testRemoveClientSubscriptions_existingClientID() { String cliendID = "FAKE_CLID_1"; - store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); + sessionsStore.addNewSubscription(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); + store.add(new ClientTopicCouple(cliendID, "finance/#")); //Exercise store.removeForClient(cliendID); @@ -267,8 +293,9 @@ public class SubscriptionsStoreTest { @Test public void testRemoveClientSubscriptions_notexistingClientID() { - String cliendID = "FAKE_CLID_1"; - store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); + String clientID = "FAKE_CLID_1"; + sessionsStore.addNewSubscription(new Subscription(clientID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false)); + store.add(new ClientTopicCouple(clientID, "finance/#")); //Exercise store.removeForClient("FAKE_CLID_2"); @@ -280,14 +307,14 @@ public class SubscriptionsStoreTest { @Test public void testOverlappingSubscriptions() { Subscription genericSub = new Subscription("FAKE_CLI_ID_1", "a/+", AbstractMessage.QOSType.EXACTLY_ONCE, false); - store.add(genericSub); + sessionsStore.addNewSubscription(genericSub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "a/+")); Subscription specificSub = new Subscription("FAKE_CLI_ID_1", "a/b", AbstractMessage.QOSType.LEAST_ONE, false); - store.add(specificSub); + sessionsStore.addNewSubscription(specificSub); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "a/b")); //Verify assertEquals(1, store.matches("a/b").size()); - //assertTrue(store.matches("/finance").contains(slashPlusSub)); - //assertFalse(store.matches("/finance").contains(anySub)); } @Test @@ -350,13 +377,15 @@ public class SubscriptionsStoreTest { ISessionsStore sessionsStore = memStore.sessionsStore(); aStore.init(sessionsStore); //subscribe a not active clientID1 to /topic - Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); - aStore.add(slashSub); sessionsStore.createNewSession("FAKE_CLI_ID_1", true).deactivate(); + Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); + sessionsStore.addNewSubscription(slashSub); + aStore.add(new ClientTopicCouple("FAKE_CLI_ID_1", "/topic")); //subscribe an active clientID2 to /topic Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", "/topic", AbstractMessage.QOSType.MOST_ONE, false); - aStore.add(slashSub2); + sessionsStore.addNewSubscription(slashSub2); + aStore.add(new ClientTopicCouple("FAKE_CLI_ID_2", "/topic")); //Exercise aStore.removeSubscription("/topic", slashSub2.getClientId()); @@ -365,13 +394,17 @@ public class SubscriptionsStoreTest { Subscription remainedSubscription = aStore.matches("/topic").get(0); assertEquals(slashSub.getClientId(), remainedSubscription.getClientId()); } - + + //TODO this test has to be moved on ClientSession, it's not part of subscription logic @Test public void overridingSubscriptions() { + ClientSession session = sessionsStore.createNewSession("FAKE_CLI_ID_1", true); Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); - store.add(oldSubscription); + session.subscribe("/topic", oldSubscription); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "/topic")); Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE, false); - store.add(overrindingSubscription); + session.subscribe("/topic", overrindingSubscription); + store.add(new ClientTopicCouple("FAKE_CLI_ID_1", "/topic")); //Verify List subscriptions = store.matches("/topic"); @@ -385,13 +418,18 @@ public class SubscriptionsStoreTest { * */ @Test public void duplicatedSubscriptionsWithDifferentQos() { + ClientSession session2 = sessionsStore.createNewSession("client2", true); Subscription client2Sub = new Subscription("client2", "client/test/b", AbstractMessage.QOSType.MOST_ONE, true); - store.add(client2Sub); + session2.subscribe("client/test/b", client2Sub); + store.add(new ClientTopicCouple("client2", "client/test/b")); + ClientSession session1 = sessionsStore.createNewSession("client1", true); Subscription client1SubQoS0 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.MOST_ONE, true); - store.add(client1SubQoS0); + session1.subscribe("client/test/b", client1SubQoS0); + store.add(new ClientTopicCouple("client1", "client/test/b")); Subscription client1SubQoS2 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.EXACTLY_ONCE, true); - store.add(client1SubQoS2); + session1.subscribe("client/test/b", client1SubQoS2); + store.add(new ClientTopicCouple("client1", "client/test/b")); System.out.println(store.dumpTree()); @@ -399,7 +437,10 @@ public class SubscriptionsStoreTest { List subscriptions = store.matches("client/test/b"); assertTrue(subscriptions.contains(client1SubQoS2)); assertTrue(subscriptions.contains(client2Sub)); - assertFalse(subscriptions.contains(client1SubQoS0)); //client1SubQoS2 should override client1SubQoS0 + + Subscription client1Sub = subscriptions.get(subscriptions.indexOf(client1SubQoS0)); + assertFalse(client1SubQoS0.getRequestedQos().equals(client1Sub.getRequestedQos())); + assertEquals(client1SubQoS2.getRequestedQos(), client1Sub.getRequestedQos()); //client1SubQoS2 should override client1SubQoS0 } @Test 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 deleted file mode 100644 index d733c04e..00000000 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TreeNodeTest.java +++ /dev/null @@ -1,33 +0,0 @@ -package io.moquette.spi.impl.subscriptions; - -import static io.moquette.proto.messages.AbstractMessage.QOSType.*; -import static junit.framework.Assert.assertEquals; - -import org.junit.Test; - -import java.util.Set; - -public class TreeNodeTest { - - public static final String CLIENT_ID = "FAKE_CLIENT_ID"; - - @Test - public void testAddSubscriptionOverwriteAnExistingWithDifferentCleanSessionFlag() throws Exception { - TreeNode root = new TreeNode(); - Subscription existingSub = new Subscription(CLIENT_ID, "/topic", LEAST_ONE, false); - root.addSubscription(existingSub); - Set subs = root.findAllByClientID(CLIENT_ID); - assertEquals(1, subs.size()); - - //Exercise - Subscription overwritingSub = new Subscription(CLIENT_ID, "/topic", LEAST_ONE, true); - root.addSubscription(overwritingSub); - - //Verify - //the second subscription has overwritten the first one - subs = root.findAllByClientID(CLIENT_ID); - assertEquals(1, subs.size()); - Subscription storedSub = subs.iterator().next(); - assertEquals(overwritingSub.isCleanSession(), storedSub.isCleanSession()); - } -} \ No newline at end of file diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java index ef7e2c52..877c4e1f 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -20,8 +20,10 @@ import static io.moquette.commons.Constants.*; import io.moquette.server.IntegrationUtils; 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.proto.messages.AbstractMessage; import io.moquette.spi.impl.subscriptions.Subscription; import org.junit.After; @@ -67,15 +69,16 @@ public class MapDBPersistentStoreTest { @Test public void overridingSubscriptions() { + ClientSession session1 = m_sessionsStore.createNewSession("FAKE_CLI_ID_1", true); Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); - m_sessionsStore.addNewSubscription(oldSubscription); + session1.subscribe("/topic", oldSubscription); Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE, false); - m_sessionsStore.addNewSubscription(overrindingSubscription); + session1.subscribe("/topic", overrindingSubscription); //Verify - List subscriptions = m_sessionsStore.listAllSubscriptions(); + List subscriptions = m_sessionsStore.listAllSubscriptions(); assertEquals(1, subscriptions.size()); - Subscription sub = subscriptions.get(0); + Subscription sub = m_sessionsStore.getSubscription(subscriptions.get(0)); assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos()); } diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index ed1e487a..cd00ce77 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -86,10 +86,9 @@ public class MemorySessionStore implements ISessionsStore { } Set subs = m_persistentSubscriptions.get(clientID); - if (!subs.contains(newSubscription)) { - subs.add(newSubscription); - m_persistentSubscriptions.put(clientID, subs); - } + subs.remove(newSubscription); //same topic and clientID + subs.add(newSubscription); + m_persistentSubscriptions.put(clientID, subs); } @Override @@ -97,11 +96,6 @@ public class MemorySessionStore implements ISessionsStore { m_persistentSubscriptions.remove(clientID); } - @Override - public void updateSubscriptions(String clientID, Set subscriptions) { - m_persistentSubscriptions.put(clientID, subscriptions); - } - @Override public boolean contains(String clientID) { return m_persistentSubscriptions.containsKey(clientID); @@ -135,14 +129,30 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public List listAllSubscriptions() { - List allSubscriptions = new ArrayList<>(); + public List listAllSubscriptions() { + List allSubscriptions = new ArrayList<>(); for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { - allSubscriptions.addAll(entry.getValue()); + for (Subscription sub : entry.getValue()) { + allSubscriptions.add(new ClientTopicCouple(sub.getClientId(), sub.getTopicFilter())); + } } return allSubscriptions; } + @Override + public Subscription getSubscription(ClientTopicCouple couple) { + Set subscriptions = m_persistentSubscriptions.get(couple.clientID); + if (subscriptions == null || subscriptions.isEmpty()) { + return null; + } + for (Subscription sub : subscriptions) { + if (sub.getTopicFilter().equals(couple.topicFilter)) { + return sub; + } + } + return null; + } + @Override public void activate(String clientID) { activationHelper(clientID, true); diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 4c9235bd..390ba3a9 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -1,25 +1,29 @@ # Set root logger level to DEBUG and its only appender to A1. log4j.rootLogger=ERROR, stdout, file -#log4j.logger.io.moquette=WARN -log4j.logger.io.moquette=DEBUG +log4j.logger.io.moquette=WARN +#log4j.logger.io.moquette=DEBUG -log4j.logger.io.moquette.spi.impl.ProtocolProcessor=DEBUG -log4j.logger.io.moquette.spi.impl.SimpleMessaging=DEBUG +log4j.logger.io.moquette.spi.impl.ProtocolProcessor=TRACE +#log4j.logger.io.moquette.spi.impl.ProtocolProcessor=WARN +#log4j.logger.io.moquette.spi.impl.SimpleMessaging=DEBUG #Protocol parsing -log4j.logger.io.moquette.server.netty.NettyMQTTHandler=DEBUG -log4j.logger.io.moquette.parser.netty=DEBUG +#log4j.logger.io.moquette.server.netty.NettyMQTTHandler=DEBUG +log4j.logger.io.moquette.server.netty.NettyMQTTHandler=WARN +#log4j.logger.io.moquette.parser.netty=DEBUG #Storage service #log4j.logger.io.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG -log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=DEBUG +#log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=DEBUG +log4j.logger.io.moquette.spi.persistence.MapDBPersistentStore=WARN +log4j.logger.io.moquette.spi.persistence.MapDBSessionsStore=DEBUG # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender #log4j.appender.stdout.Threshold=WARN #log4j.appender.stdout.Threshold=INFO -log4j.appender.stdout.Threshold=DEBUG +log4j.appender.stdout.Threshold=TRACE # for debug trace #log4j.appender.stdout.Threshold=TRACE log4j.appender.stdout.layout=org.apache.log4j.PatternLayout