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

Architechtural refactory, moved subscription out of tree and into db store

Browse Source
This commit is contained in:
andsel
2015-12-06 11:43:29 +01:00
parent 490c6a763d
commit 9ce9a6597f
15 changed files with 260 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+1 -1
View File
@@ -98,7 +98,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
}
m_processor.processConnectionLost(clientID, stolen);
}
ctx.close(/*false*/);
ctx.close();
}
@Override
broker/src/main/java/io/moquette/spi/ClientSession.java
+10 -2
View File
@@ -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;
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+43 -4
View File
@@ -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<Subscription> subscriptions);
List<ClientTopicCouple> listAllSubscriptions();
List<Subscription> 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
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+8 -5
View File
@@ -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<IMessagesStore.StoredMessage> 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());
broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java
+6 -7
View File
@@ -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;
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+17 -26
View File
@@ -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<TreeNode> 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<Subscription> subscriptions = sessionsStore.listAllSubscriptions();
m_sessionsStore = sessionsStore;
List<ClientTopicCouple> 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<Token> tokenQueue = new LinkedBlockingDeque<>(tokens);
List<Subscription> matchingSubs = new ArrayList<>();
List<ClientTopicCouple> matchingSubs = new ArrayList<>();
subscriptions.get().matches(tokenQueue, matchingSubs);
//remove the overlapping subscriptions, selecting ones with greatest qos
Map<String, Subscription> 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());
broker/src/main/java/io/moquette/spi/impl/subscriptions/TreeNode.java
+16 -46
View File
@@ -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<Subscription> {
public int compare(Subscription o1, Subscription o2) {
return o1.getClientId().compareTo(o2.getClientId());
}
}
Token m_token;
List<TreeNode> m_children = new ArrayList<>();
List<Subscription> m_subscriptions = new ArrayList<>();
//TODO move to set of ClientIDthe set of clientIDs that has subscriptions to this topic
Set<ClientTopicCouple> 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<Subscription> 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<Subscription> subscriptions() {
Collection<ClientTopicCouple> subscriptions() {
return m_subscriptions;
}
void matches(Queue<Token> tokens, List<Subscription> matchingSubs) {
public void remove(ClientTopicCouple clientTopicCouple) {
m_subscriptions.remove(clientTopicCouple);
}
void matches(Queue<Token> tokens, List<ClientTopicCouple> 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<Subscription> subsToRemove = new ArrayList<>();
for (Subscription s : newSubRoot.m_subscriptions) {
if (s.clientId.equals(clientID)) {
List<ClientTopicCouple> 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<Subscription> findAllByClientID(String clientID) {
Set<Subscription> 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;
}
}
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+30 -56
View File
@@ -41,8 +41,6 @@ class MapDBSessionsStore implements ISessionsStore {
private ConcurrentMap<String, Map<Integer, String>> m_inflightStore;
//map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> m_inFlightIds;
//persistent Map of clientID, set of Subscriptions
private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
//maps clientID->[guid*], insertion order cares, it's queue
private ConcurrentMap<String, List<String>> 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<Subscription>());
}
m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription);
Set<Subscription> 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<Subscription> 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<Subscription> subscriptions) {
m_persistentSubscriptions.put(clientID, subscriptions);
}
@Override
public List<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<>();
for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
allSubscriptions.addAll(entry.getValue());
public List<ClientTopicCouple> listAllSubscriptions() {
final List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
for (String clientID : m_persistentSessions.keySet()) {
ConcurrentMap<String, Subscription> 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<String, Subscription> 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<Subscription>());
m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false));
return new ClientSession(clientID, m_messagesStore, this, cleanSession);
}
broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java
+3 -5
View File
@@ -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()));
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+6 -3
View File
@@ -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() {
broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java
+79 -38
View File
@@ -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<Subscription> 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<Subscription> 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
broker/src/test/java/io/moquette/spi/impl/subscriptions/TreeNodeTest.java
-33
View File
@@ -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<Subscription> 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());
}
}
broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java
+7 -4
View File
@@ -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<Subscription> subscriptions = m_sessionsStore.listAllSubscriptions();
List<ClientTopicCouple> 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());
}
broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java
+22 -12
View File
@@ -86,10 +86,9 @@ public class MemorySessionStore implements ISessionsStore {
}
Set<Subscription> 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<Subscription> 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<Subscription> listAllSubscriptions() {
List<Subscription> allSubscriptions = new ArrayList<>();
public List<ClientTopicCouple> listAllSubscriptions() {
List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
for (Map.Entry<String, Set<Subscription>> 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<Subscription> 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);
broker/src/test/resources/log4j.properties
+12 -8
View File
@@ -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