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

Changed SubscriptionsStore overall tree (ones based on clientID) operations to copy all the tree

Browse Source
This commit is contained in:
andsel
2015-10-10 19:16:30 +02:00
parent ef9af6edae
commit 85872eec19
4 changed files with 107 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+2 -1
View File
@@ -18,7 +18,6 @@ package org.eclipse.moquette.spi.impl;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import org.eclipse.moquette.server.netty.NettyChannel;
import org.eclipse.moquette.spi.IMatchingCondition;
@@ -427,6 +426,8 @@ public class ProtocolProcessor {
}
LOG.debug("clientIDs are {}", m_clientIDs);
if (m_clientIDs.get(clientId) == null) {
//TODO while we were publishing to the target client, that client disconnected,
// could happen is not an error HANDLE IT
throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", clientId, m_clientIDs));
}
ServerChannel session = m_clientIDs.get(clientId).getSession();
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java
+9 -1
View File
@@ -39,7 +39,15 @@ public class Subscription implements Serializable {
this.topicFilter = topicFilter;
this.cleanSession = cleanSession;
}
public Subscription(Subscription orig) {
this.requestedQos = orig.requestedQos;
this.clientId = orig.clientId;
this.topicFilter = orig.topicFilter;
this.cleanSession = orig.cleanSession;
this.active = orig.active;
}
/**
* Factory method for empty subscriptions
*/
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+64 -50
View File
@@ -64,6 +64,7 @@ public class SubscriptionsStore {
String s = "";
@Override
public void visit(TreeNode node, int deep) {
String subScriptionsStr = "";
String indentTabs = indentTabs(deep);
@@ -82,7 +83,8 @@ public class SubscriptionsStore {
}
return s;
}
@Override
public String getResult() {
return s;
}
@@ -90,12 +92,14 @@ public class SubscriptionsStore {
private class SubscriptionTreeCollector implements IVisitor<List<Subscription>> {
private List<Subscription> m_allSubscriptions = new ArrayList<Subscription>();
private List<Subscription> m_allSubscriptions = new ArrayList<>();
@Override
public void visit(TreeNode node, int deep) {
m_allSubscriptions.addAll(node.subscriptions());
}
@Override
public List<Subscription> getResult() {
return m_allSubscriptions;
}
@@ -136,32 +140,9 @@ public class SubscriptionsStore {
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
} while(!subscriptions.compareAndSet(oldRoot, couple.root));
LOG.debug("root ref {}, original root was {}", couple.root, oldRoot);
}
private TreeNode findMatchingNode(String topic) {
List<Token> tokens = new ArrayList<>();
try {
tokens = parseTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
LOG.error(null, ex);
// return;
}
TreeNode current = subscriptions.get();
for (Token token : tokens) {
TreeNode matchingChildren;
//check if a children with the same token already exists
if ((matchingChildren = current.childWithToken(token)) != null) {
current = matchingChildren;
} else {
//can't find any match
return null;
}
}
return current;
}
protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) {
List<Token> tokens = new ArrayList<>();
@@ -203,20 +184,27 @@ public class SubscriptionsStore {
public void removeSubscription(String topic, String clientID) {
TreeNode matchNode = findMatchingNode(topic);
//search for the subscription to remove
Subscription toBeRemoved = null;
for (Subscription sub : matchNode.subscriptions()) {
if (sub.topicFilter.equals(topic) && sub.getClientId().equals(clientID)) {
toBeRemoved = sub;
break;
TreeNode oldRoot;
NodeCouple couple;
do {
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) {
matchNode.subscriptions().remove(toBeRemoved);
}
if (toBeRemoved != null) {
couple.createdNode.subscriptions().remove(toBeRemoved);
}
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, couple.root));
}
/**
@@ -228,35 +216,61 @@ public class SubscriptionsStore {
List<Subscription> allSubscriptions = subsCollector.getResult();
for (Subscription subscription : allSubscriptions) {
//TODO modify the data structure
removeSubscription(subscription.getTopicFilter(), subscription.getClientId());
}
}
/**
* Visit the topics tree to remove matching subscriptions with clientID
* Visit the topics tree to remove matching subscriptions with clientID.
* It's a mutating structure operation so create a new subscription tree (partial or total).
*/
//TODO modify the data structure
public void removeForClient(String clientID) {
subscriptions.get().removeClientSubscriptions(clientID);
TreeNode oldRoot;
TreeNode newRoot;
do {
oldRoot = subscriptions.get();
newRoot = oldRoot.removeClientSubscriptions(clientID);
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, newRoot));
//persist the update
m_sessionsStore.wipeSubscriptions(clientID);
}
//TODO modify the data structure
/**
* Visit the topics tree to deactivate matching subscriptions with clientID.
* It's a mutating structure operation so create a new subscription tree (partial or total).
*/
public void deactivate(String clientID) {
subscriptions.get().deactivate(clientID);
LOG.debug("Disactivating subscriptions for clientID <{}>", clientID);
TreeNode oldRoot;
TreeNode newRoot;
do {
oldRoot = subscriptions.get();
newRoot = oldRoot.deactivate(clientID);
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, newRoot));
//persist the update
Set<Subscription> subs = subscriptions.get().findAllByClientID(clientID);
Set<Subscription> subs = newRoot.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
//TODO modify the data structure
/**
* Visit the topics tree to activate matching subscriptions with clientID.
* It's a mutating structure operation so create a new subscription tree (partial or total).
*/
public void activate(String clientID) {
LOG.debug("Activating subscriptions for clientID <{}>", clientID);
subscriptions.get().activate(clientID);
TreeNode oldRoot;
TreeNode newRoot;
do {
oldRoot = subscriptions.get();
newRoot = oldRoot.activate(clientID);
//spin lock repeating till we can, swap root, if can't swap just re-do the operation
} while(!subscriptions.compareAndSet(oldRoot, newRoot));
//persist the update
Set<Subscription> subs = subscriptions.get().findAllByClientID(clientID);
Set<Subscription> subs = newRoot.findAllByClientID(clientID);
m_sessionsStore.updateSubscriptions(clientID, subs);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java
+32 -14
View File
@@ -73,7 +73,10 @@ class TreeNode {
final TreeNode copy = new TreeNode(this);
copy.m_parent = m_parent;
copy.m_children = new ArrayList<>(m_children);
copy.m_subscriptions = new ArrayList<>(m_subscriptions);
copy.m_subscriptions = new ArrayList<>(m_subscriptions.size());
for (Subscription sub : m_subscriptions) {
copy.m_subscriptions.add(new Subscription(sub));
}
copy.m_token = m_token;
return copy;
}
@@ -144,55 +147,70 @@ class TreeNode {
return res;
}
void removeClientSubscriptions(String clientID) {
/**
* Create a copied subtree rooted on this node but purged of clientID's subscriptions.
* */
TreeNode removeClientSubscriptions(String clientID) {
//collect what to delete and then delete to avoid ConcurrentModification
TreeNode newSubRoot = this.copy();
List<Subscription> subsToRemove = new ArrayList<>();
for (Subscription s : m_subscriptions) {
for (Subscription s : newSubRoot.m_subscriptions) {
if (s.clientId.equals(clientID)) {
subsToRemove.add(s);
}
}
for (Subscription s : subsToRemove) {
m_subscriptions.remove(s);
newSubRoot.m_subscriptions.remove(s);
}
//go deep
for (TreeNode child : m_children) {
child.removeClientSubscriptions(clientID);
List<TreeNode> newChildren = new ArrayList<>(newSubRoot.m_children.size());
for (TreeNode child : newSubRoot.m_children) {
newChildren.add(child.removeClientSubscriptions(clientID));
}
newSubRoot.m_children = newChildren;
return newSubRoot;
}
/**
* Deactivate all topic subscriptions for the given clientID.
* */
void deactivate(String clientID) {
for (Subscription s : m_subscriptions) {
TreeNode deactivate(String clientID) {
TreeNode newSubRoot = this.copy();
for (Subscription s : newSubRoot.m_subscriptions) {
if (s.clientId.equals(clientID)) {
s.setActive(false);
}
}
//go deep
for (TreeNode child : m_children) {
child.deactivate(clientID);
List<TreeNode> newChildren = new ArrayList<>(newSubRoot.m_children.size());
for (TreeNode child : newSubRoot.m_children) {
newChildren.add(child.deactivate(clientID));
}
newSubRoot.m_children = newChildren;
return newSubRoot;
}
/**
* Activate all topic subscriptions for the given clientID.
* */
public void activate(String clientID) {
for (Subscription s : m_subscriptions) {
TreeNode activate(String clientID) {
TreeNode newSubRoot = this.copy();
for (Subscription s : newSubRoot.m_subscriptions) {
if (s.clientId.equals(clientID)) {
s.setActive(true);
}
}
//go deep
for (TreeNode child : m_children) {
child.activate(clientID);
List<TreeNode> newChildren = new ArrayList<>(newSubRoot.m_children.size());
for (TreeNode child : newSubRoot.m_children) {
newChildren.add(child.activate(clientID));
}
newSubRoot.m_children = newChildren;
return newSubRoot;
}
/**