anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

broker/SubscriptionsStore: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:41 +01:00
parent fdf5a83386
commit 4db4663274
1 changed files with 115 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+92 -59
View File
@@ -13,13 +13,13 @@
* *
* You may elect to redistribute this code under either of these licenses. * You may elect to redistribute this code under either of these licenses.
*/ */
package io.moquette.spi.impl.subscriptions; package io.moquette.spi.impl.subscriptions;
import io.moquette.spi.ISessionsStore; import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple; import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.text.ParseException; import java.text.ParseException;
import java.util.*; import java.util.*;
import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingDeque;
@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicReference;
public class SubscriptionsStore { public class SubscriptionsStore {
public static class NodeCouple { public static class NodeCouple {
final TreeNode root; final TreeNode root;
final TreeNode createdNode; final TreeNode createdNode;
@@ -44,21 +45,27 @@ public class SubscriptionsStore {
/** /**
* Check if the topic filter of the subscription is well formed * Check if the topic filter of the subscription is well formed
* @param topicFilter the filter to validate *
* @param topicFilter
* the filter to validate
* @return true if it's correct. * @return true if it's correct.
* */ */
public static boolean validate(String topicFilter) { public static boolean validate(String topicFilter) {
try { try {
parseTopic(topicFilter); parseTopic(topicFilter);
return true; return true;
} catch (ParseException pex) { } catch (ParseException pex) {
LOG.warn("The topic filter is malformed. TopicFilter = {}, cause = {}, errorMessage = {}.", topicFilter, LOG.warn(
pex.getCause(), pex.getMessage()); "The topic filter is malformed. TopicFilter = {}, cause = {}, errorMessage = {}.",
topicFilter,
pex.getCause(),
pex.getMessage());
return false; return false;
} }
} }
public interface IVisitor<T> { public interface IVisitor<T> {
void visit(TreeNode node, int deep); void visit(TreeNode node, int deep);
T getResult(); T getResult();
@@ -76,14 +83,15 @@ public class SubscriptionsStore {
subScriptionsStr += indentTabs + couple.toString() + "\n"; subScriptionsStr += indentTabs + couple.toString() + "\n";
} }
s += node.getToken() == null ? "" : node.getToken().toString(); s += node.getToken() == null ? "" : node.getToken().toString();
s += "\n" + (node.m_subscriptions.isEmpty() ? indentTabs : "") + subScriptionsStr /*+ "\n"*/; s += "\n" + (node.m_subscriptions.isEmpty() ? indentTabs : "")
+ subScriptionsStr /* + "\n" */;
} }
private String indentTabs(int deep) { private String indentTabs(int deep) {
String s = ""; String s = "";
for (int i=0; i < deep; i++) { for (int i = 0; i < deep; i++) {
s += "\t"; s += "\t";
// s += "--"; // s += "--";
} }
return s; return s;
} }
@@ -99,21 +107,26 @@ public class SubscriptionsStore {
private volatile ISessionsStore m_sessionsStore; private volatile ISessionsStore m_sessionsStore;
/** /**
* Initialize the subscription tree with the list of subscriptions. * Initialize the subscription tree with the list of subscriptions. Maintained for compatibility
* Maintained for compatibility reasons. * reasons.
* @param sessionsStore to be used as backing store from the subscription store. *
* @param sessionsStore
* to be used as backing store from the subscription store.
*/ */
public void init(ISessionsStore sessionsStore) { public void init(ISessionsStore sessionsStore) {
LOG.info("Initializing subscriptions store..."); LOG.info("Initializing subscriptions store...");
m_sessionsStore = sessionsStore; m_sessionsStore = sessionsStore;
List<ClientTopicCouple> subscriptions = sessionsStore.listAllSubscriptions(); List<ClientTopicCouple> subscriptions = sessionsStore.listAllSubscriptions();
//reload any subscriptions persisted // reload any subscriptions persisted
if (LOG.isTraceEnabled()) { if (LOG.isTraceEnabled()) {
LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}.", dumpTree()); LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}.", dumpTree());
} }
for (ClientTopicCouple clientTopic : subscriptions) { for (ClientTopicCouple clientTopic : subscriptions) {
LOG.info("Re-subscribing client to topic. ClientId = {}, topicFilter = {}.", clientTopic.clientID, clientTopic.topicFilter); LOG.info(
"Re-subscribing client to topic. ClientId = {}, topicFilter = {}.",
clientTopic.clientID,
clientTopic.topicFilter);
add(clientTopic); add(clientTopic);
} }
if (LOG.isTraceEnabled()) { if (LOG.isTraceEnabled()) {
@@ -123,27 +136,30 @@ public class SubscriptionsStore {
public void add(ClientTopicCouple newSubscription) { public void add(ClientTopicCouple newSubscription) {
/* /*
* The topic filters have already been validated at the ProtocolProcessor. We can assume they are valid. * The topic filters have already been validated at the ProtocolProcessor. We can assume
* they are valid.
*/ */
TreeNode oldRoot; TreeNode oldRoot;
NodeCouple couple; NodeCouple couple;
do { do {
oldRoot = subscriptions.get(); oldRoot = subscriptions.get();
couple = recreatePath(newSubscription.topicFilter, oldRoot); couple = recreatePath(newSubscription.topicFilter, oldRoot);
couple.createdNode.addSubscription(newSubscription); //createdNode could be null? couple.createdNode.addSubscription(newSubscription); // createdNode could be null?
//spin lock repeating till we can, swap root, if can't swap just re-do the operation // spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, couple.root)); } while (!subscriptions.compareAndSet(oldRoot, couple.root));
LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot); LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot);
} }
protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) { protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) {
List<Token> tokens; List<Token> tokens;
try { try {
tokens = parseTopic(topic); tokens = parseTopic(topic);
} catch (ParseException ex) { } catch (ParseException ex) {
//TODO handle the parse exception // TODO handle the parse exception
LOG.error("The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.", topic, ex.getCause(), LOG.error(
"The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.",
topic,
ex.getCause(),
ex.getMessage()); ex.getMessage());
throw new IllegalArgumentException(ex.getMessage()); throw new IllegalArgumentException(ex.getMessage());
} }
@@ -154,15 +170,15 @@ public class SubscriptionsStore {
for (Token token : tokens) { for (Token token : tokens) {
TreeNode matchingChildren; TreeNode matchingChildren;
//check if a children with the same token already exists // check if a children with the same token already exists
if ((matchingChildren = current.childWithToken(token)) != null) { if ((matchingChildren = current.childWithToken(token)) != null) {
//copy the traversed node // copy the traversed node
current = matchingChildren.copy(); current = matchingChildren.copy();
//update the child just added in the children list // update the child just added in the children list
parent.updateChild(matchingChildren, current); parent.updateChild(matchingChildren, current);
parent = current; parent = current;
} else { } else {
//create a new node for the newly inserted token // create a new node for the newly inserted token
matchingChildren = new TreeNode(); matchingChildren = new TreeNode();
matchingChildren.setToken(token); matchingChildren.setToken(token);
current.addChild(matchingChildren); current.addChild(matchingChildren);
@@ -174,7 +190,8 @@ public class SubscriptionsStore {
public void removeSubscription(String topic, String clientID) { public void removeSubscription(String topic, String clientID) {
/* /*
* The topic filters have already been validated at the ProtocolProcessor. We can assume they are valid. * The topic filters have already been validated at the ProtocolProcessor. We can assume
* they are valid.
*/ */
TreeNode oldRoot; TreeNode oldRoot;
NodeCouple couple; NodeCouple couple;
@@ -183,14 +200,16 @@ public class SubscriptionsStore {
couple = recreatePath(topic, oldRoot); couple = recreatePath(topic, oldRoot);
couple.createdNode.remove(new ClientTopicCouple(clientID, topic)); couple.createdNode.remove(new ClientTopicCouple(clientID, topic));
//spin lock repeating till we can, swap root, if can't swap just re-do the operation // spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, couple.root)); } while (!subscriptions.compareAndSet(oldRoot, couple.root));
} }
/** /**
* Visit the topics tree to remove matching subscriptions with clientID. * Visit the topics tree to remove matching subscriptions with clientID. It's a mutating
* It's a mutating structure operation so create a new subscription tree (partial or total). * structure operation so create a new subscription tree (partial or total).
* @param clientID the client ID to remove. *
* @param clientID
* the client ID to remove.
*/ */
public void removeForClient(String clientID) { public void removeForClient(String clientID) {
TreeNode oldRoot; TreeNode oldRoot;
@@ -198,17 +217,17 @@ public class SubscriptionsStore {
do { do {
oldRoot = subscriptions.get(); oldRoot = subscriptions.get();
newRoot = oldRoot.removeClientSubscriptions(clientID); newRoot = oldRoot.removeClientSubscriptions(clientID);
//spin lock repeating till we can, swap root, if can't swap just re-do the operation // spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, newRoot)); } while (!subscriptions.compareAndSet(oldRoot, newRoot));
} }
/** /**
* Given a topic string return the clients subscriptions that matches it. * Given a topic string return the clients subscriptions that matches it. Topic string can't
* Topic string can't contain character # and + because they are reserved to * contain character # and + because they are reserved to listeners subscriptions, and not topic
* listeners subscriptions, and not topic publishing. * publishing.
* *
* @param topic to use fo searching matching subscriptions. * @param topic
* to use fo searching matching subscriptions.
* @return the list of matching subscriptions, or empty if not matching. * @return the list of matching subscriptions, or empty if not matching.
*/ */
public List<Subscription> matches(String topic) { public List<Subscription> matches(String topic) {
@@ -216,8 +235,12 @@ public class SubscriptionsStore {
try { try {
tokens = parseTopic(topic); tokens = parseTopic(topic);
} catch (ParseException ex) { } catch (ParseException ex) {
//TODO handle the parse exception // TODO handle the parse exception
LOG.warn("The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.", topic, ex.getCause(), ex.getMessage()); LOG.warn(
"The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.",
topic,
ex.getCause(),
ex.getMessage());
return Collections.emptyList(); return Collections.emptyList();
} }
@@ -225,16 +248,16 @@ public class SubscriptionsStore {
List<ClientTopicCouple> matchingSubs = new ArrayList<>(); List<ClientTopicCouple> matchingSubs = new ArrayList<>();
subscriptions.get().matches(tokenQueue, matchingSubs); subscriptions.get().matches(tokenQueue, matchingSubs);
//remove the overlapping subscriptions, selecting ones with greatest qos // remove the overlapping subscriptions, selecting ones with greatest qos
Map<String, Subscription> subsForClient = new HashMap<>(); Map<String, Subscription> subsForClient = new HashMap<>();
for (ClientTopicCouple matchingCouple : matchingSubs) { for (ClientTopicCouple matchingCouple : matchingSubs) {
Subscription existingSub = subsForClient.get(matchingCouple.clientID); Subscription existingSub = subsForClient.get(matchingCouple.clientID);
Subscription sub = m_sessionsStore.getSubscription(matchingCouple); Subscription sub = m_sessionsStore.getSubscription(matchingCouple);
if (sub == null) { if (sub == null) {
//if the m_sessionStore hasn't the sub because the client disconnected // if the m_sessionStore hasn't the sub because the client disconnected
continue; continue;
} }
//update the selected subscriptions if not present or if has a greater qos // update the selected subscriptions if not present or if has a greater qos
if (existingSub == null || existingSub.getRequestedQos().value() < sub.getRequestedQos().value()) { if (existingSub == null || existingSub.getRequestedQos().value() < sub.getRequestedQos().value()) {
subsForClient.put(matchingCouple.clientID, sub); subsForClient.put(matchingCouple.clientID, sub);
} }
@@ -268,17 +291,20 @@ public class SubscriptionsStore {
/** /**
* Verify if the 2 topics matching respecting the rules of MQTT Appendix A * Verify if the 2 topics matching respecting the rules of MQTT Appendix A
* @param msgTopic the topic to match from the message *
* @param subscriptionTopic the topic filter of the subscription * @param msgTopic
* the topic to match from the message
* @param subscriptionTopic
* the topic filter of the subscription
* @return true if the two topics match. * @return true if the two topics match.
*/ */
//TODO reimplement with iterators or with queues // TODO reimplement with iterators or with queues
public static boolean matchTopics(String msgTopic, String subscriptionTopic) { public static boolean matchTopics(String msgTopic, String subscriptionTopic) {
try { try {
List<Token> msgTokens = SubscriptionsStore.parseTopic(msgTopic); List<Token> msgTokens = SubscriptionsStore.parseTopic(msgTopic);
List<Token> subscriptionTokens = SubscriptionsStore.parseTopic(subscriptionTopic); List<Token> subscriptionTokens = SubscriptionsStore.parseTopic(subscriptionTopic);
int i = 0; int i = 0;
for (; i< subscriptionTokens.size(); i++) { for (; i < subscriptionTokens.size(); i++) {
Token subToken = subscriptionTokens.get(i); Token subToken = subscriptionTokens.get(i);
if (subToken != Token.MULTI && subToken != Token.SINGLE) { if (subToken != Token.MULTI && subToken != Token.SINGLE) {
if (i >= msgTokens.size()) { if (i >= msgTokens.size()) {
@@ -293,19 +319,23 @@ public class SubscriptionsStore {
return true; return true;
} }
if (subToken == Token.SINGLE) { if (subToken == Token.SINGLE) {
//skip a step forward // skip a step forward
} }
} }
} }
//if last token was a SINGLE then treat it as an empty // if last token was a SINGLE then treat it as an empty
// if (subToken == Token.SINGLE && (i - msgTokens.size() == 1)) { // if (subToken == Token.SINGLE && (i - msgTokens.size() == 1)) {
// i--; // i--;
// } // }
return i == msgTokens.size(); return i == msgTokens.size();
} catch (ParseException ex) { } catch (ParseException ex) {
LOG.error( LOG.error(
"The message topic, the subscription topic or both are malformed. MsgTopic = {}, subscriptionTopic = {}, cause = {}, errorMessage = {}.", "The message topic, the subscription topic or both are malformed. "
msgTopic, subscriptionTopic, ex.getCause(), ex.getMessage()); + "MsgTopic = {}, subscriptionTopic = {}, cause = {}, errorMessage = {}.",
msgTopic,
subscriptionTopic,
ex.getCause(),
ex.getMessage());
throw new IllegalStateException(ex.getMessage()); throw new IllegalStateException(ex.getMessage());
} }
} }
@@ -319,7 +349,7 @@ public class SubscriptionsStore {
} }
if (topic.endsWith("/")) { if (topic.endsWith("/")) {
//Add a fictious space // Add a fictious space
String[] newSplitted = new String[splitted.length + 1]; String[] newSplitted = new String[splitted.length + 1];
System.arraycopy(splitted, 0, newSplitted, 0, splitted.length); System.arraycopy(splitted, 0, newSplitted, 0, splitted.length);
newSplitted[splitted.length] = ""; newSplitted[splitted.length] = "";
@@ -329,14 +359,17 @@ public class SubscriptionsStore {
for (int i = 0; i < splitted.length; i++) { for (int i = 0; i < splitted.length; i++) {
String s = splitted[i]; String s = splitted[i];
if (s.isEmpty()) { if (s.isEmpty()) {
// if (i != 0) { // if (i != 0) {
// throw new ParseException("Bad format of topic, expetec topic name between separators", i); // throw new ParseException("Bad format of topic, expetec topic name between
// } // separators", i);
// }
res.add(Token.EMPTY); res.add(Token.EMPTY);
} else if (s.equals("#")) { } else if (s.equals("#")) {
//check that multi is the last symbol // check that multi is the last symbol
if (i != splitted.length - 1) { if (i != splitted.length - 1) {
throw new ParseException("Bad format of topic, the multi symbol (#) has to be the last one after a separator", i); throw new ParseException(
"Bad format of topic, the multi symbol (#) has to be the last one after a separator",
i);
} }
res.add(Token.MULTI); res.add(Token.MULTI);
} else if (s.contains("#")) { } else if (s.contains("#")) {