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

Merge pull request #270 from AndreasMager/add-topic-class

Browse Source 
New class added: Topic
This commit is contained in:
Andrea Selva
2017-02-19 19:35:54 +01:00
committed by GitHub
parent 44fecb3b5a a2e51b2c31
commit b0095aa3fd
34 changed files with 589 additions and 448 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java
+1 -1
View File
@@ -26,7 +26,7 @@ public class InterceptSubscribeMessage implements InterceptMessage {
}
public String getTopicFilter() {
return subscription.getTopicFilter();
return subscription.getTopicFilter().toString();
}
public String getUsername() {
broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java
+1 -1
View File
@@ -147,7 +147,7 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
new MqttSubscription(
subscription.getRequestedQos().toString(),
subscription.getClientId(),
subscription.getTopicFilter(),
subscription.getTopicFilter().toString(),
subscription.isActive()));
}
result.setActiveSubscriptions(mqttSubscriptions);
broker/src/main/java/io/moquette/spi/ClientSession.java
+3 -3
View File
@@ -18,7 +18,7 @@ package io.moquette.spi;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashSet;
@@ -89,7 +89,7 @@ public class ClientSession {
newSubscription.getClientId(),
newSubscription.getTopicFilter(),
newSubscription.getRequestedQos());
boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
boolean validTopic = newSubscription.getTopicFilter().isValid();
if (!validTopic) {
LOG.warn(
"The topic filter is not valid. MqttClientId = {}, topics = {}.",
@@ -118,7 +118,7 @@ public class ClientSession {
return true;
}
public void unsubscribeFrom(String topicFilter) {
public void unsubscribeFrom(Topic topicFilter) {
LOG.info("Removing subscription. MqttClientID = {}, topics = {}.", clientID, topicFilter);
m_sessionsStore.removeSubscription(topicFilter, clientID);
Set<Subscription> subscriptionsToRemove = new HashSet<>();
broker/src/main/java/io/moquette/spi/IMatchingCondition.java
+3 -1
View File
@@ -16,9 +16,11 @@
package io.moquette.spi;
import io.moquette.spi.impl.subscriptions.Topic;
/**
*/
public interface IMatchingCondition {
boolean match(String key);
boolean match(Topic key);
}
broker/src/main/java/io/moquette/spi/IMessagesStore.java
+3 -2
View File
@@ -16,6 +16,7 @@
package io.moquette.spi;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
@@ -114,7 +115,7 @@ public interface IMessagesStore {
* @param guid
* of the message to mark as retained.
*/
void storeRetained(String topic, MessageGUID guid);
void storeRetained(Topic topic, MessageGUID guid);
/**
* Return a list of retained messages that satisfy the condition.
@@ -138,7 +139,7 @@ public interface IMessagesStore {
StoredMessage getMessageByGuid(MessageGUID guid);
void cleanRetained(String topic);
void cleanRetained(Topic topic);
int getPendingPublishMessages(String clientID);
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+4 -3
View File
@@ -21,6 +21,7 @@ import java.util.List;
import java.util.concurrent.BlockingQueue;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
/**
* Store used to handle the persistence of the subscriptions tree.
@@ -31,10 +32,10 @@ public interface ISessionsStore {
class ClientTopicCouple {
public final String topicFilter;
public final Topic topicFilter;
public final String clientID;
public ClientTopicCouple(String clientID, String topicFilter) {
public ClientTopicCouple(String clientID, Topic topicFilter) {
this.clientID = clientID;
this.topicFilter = topicFilter;
}
@@ -87,7 +88,7 @@ public interface ISessionsStore {
* @param clientID
* the session client.
*/
void removeSubscription(String topic, String clientID);
void removeSubscription(Topic topic, String clientID);
/**
* Remove all the subscriptions of the session
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+20 -15
View File
@@ -26,6 +26,7 @@ import io.moquette.spi.*;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import io.netty.buffer.ByteBuf;
@@ -610,7 +611,7 @@ public class ProtocolProcessor {
*/
public void internalPublish(MqttPublishMessage msg, final String clientId) {
final MqttQoS qos = msg.fixedHeader().qosLevel();
final String topic = msg.variableHeader().topicName();
final Topic topic = new Topic(msg.variableHeader().topicName());
LOG.info("Sending PUBLISH message. Topic = {}, qos = {}.", topic, qos);
MessageGUID guid = null;
@@ -656,7 +657,7 @@ public class ProtocolProcessor {
IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
tobeStored.setClientID(clientID);
tobeStored.setMessageID(messageId);
String topic = tobeStored.getTopic();
Topic topic = new Topic(tobeStored.getTopic());
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
LOG.info("Publishing will message. CId = {}, messageId = {}, topic = {}.", clientID, messageId, topic);
@@ -836,8 +837,9 @@ public class ProtocolProcessor {
LOG.info("Processing UNSUBSCRIBE message. CId = {}, topics = {}.", clientID, topics);
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
for (String topic : topics) {
boolean validTopic = SubscriptionsStore.validate(topic);
for (String t : topics) {
Topic topic = new Topic(t);
boolean validTopic = topic.isValid();
if (!validTopic) {
// close the connection, not valid topicFilter is a protocol violation
channel.close();
@@ -853,7 +855,7 @@ public class ProtocolProcessor {
subscriptions.removeSubscription(topic, clientID);
clientSession.unsubscribeFrom(topic);
String username = NettyUtils.userName(channel);
m_interceptor.notifyTopicUnsubscribed(topic, clientID, username);
m_interceptor.notifyTopicUnsubscribed(topic.toString(), clientID, username);
}
// ack the client
@@ -926,7 +928,9 @@ public class ProtocolProcessor {
if (req.qualityOfService() == FAILURE) {
continue;
}
Subscription newSubscription = new Subscription(clientID, req.topicName(), req.qualityOfService());
Subscription newSubscription =
new Subscription(clientID, new Topic(req.topicName()), req.qualityOfService());
clientSession.subscribe(newSubscription);
newSubscriptions.add(newSubscription);
}
@@ -948,7 +952,8 @@ public class ProtocolProcessor {
final int messageId = messageId(msg);
for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) {
if (!m_authorizator.canRead(req.topicName(), username, clientSession.clientID)) {
Topic topic = new Topic(req.topicName());
if (!m_authorizator.canRead(topic, username, clientSession.clientID)) {
// send SUBACK with 0x80, the user hasn't credentials to read the topic
LOG.error(
"The client does not have read permissions on the topic. "
@@ -956,18 +961,18 @@ public class ProtocolProcessor {
clientID,
username,
messageId,
req.topicName());
ackTopics.add(new MqttTopicSubscription(req.topicName(), FAILURE));
topic);
ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE));
} else {
MqttQoS qos;
if (SubscriptionsStore.validate(req.topicName())) {
if (topic.isValid()) {
LOG.info(
"The client will be subscribed to the topic. "
+ "CId = {}, username = {}, messageId = {}, topic = {}.",
clientID,
username,
messageId,
req.topicName());
topic);
qos = req.qualityOfService();
} else {
LOG.error(
@@ -975,10 +980,10 @@ public class ProtocolProcessor {
clientID,
username,
messageId,
req.topicName());
topic);
qos = FAILURE;
}
ackTopics.add(new MqttTopicSubscription(req.topicName(), qos));
ackTopics.add(new MqttTopicSubscription(topic.toString(), qos));
}
}
return ackTopics;
@@ -1009,8 +1014,8 @@ public class ProtocolProcessor {
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
@Override
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, newSubscription.getTopicFilter());
public boolean match(Topic key) {
return key.match(newSubscription.getTopicFilter());
}
});
broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java
+2 -1
View File
@@ -5,6 +5,7 @@ import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
@@ -33,7 +34,7 @@ class Qos0PublishHandler extends QosPublishHandler {
void receivedPublishQos0(Channel channel, MqttPublishMessage msg) {
// verify if topic can be write
final String topic = msg.variableHeader().topicName();
final Topic topic = new Topic(msg.variableHeader().topicName());
if (checkWriteOnTopic(topic, channel)) {
return;
}
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+2 -1
View File
@@ -7,6 +7,7 @@ import io.moquette.spi.IMessagesStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
@@ -43,7 +44,7 @@ class Qos1PublishHandler extends QosPublishHandler {
void receivedPublishQos1(Channel channel, MqttPublishMessage msg) {
// verify if topic can be write
final String topic = msg.variableHeader().topicName();
final Topic topic = new Topic(msg.variableHeader().topicName());
if (checkWriteOnTopic(topic, channel)) {
return;
}
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+3 -2
View File
@@ -9,6 +9,7 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
@@ -45,7 +46,7 @@ class Qos2PublishHandler extends QosPublishHandler {
}
void receivedPublishQos2(Channel channel, MqttPublishMessage msg) {
final String topic = msg.variableHeader().topicName();
final Topic topic = new Topic(msg.variableHeader().topicName());
// check if the topic can be wrote
if (checkWriteOnTopic(topic, channel)) {
return;
@@ -102,7 +103,7 @@ class Qos2PublishHandler extends QosPublishHandler {
LOG.info("Processing PUBREL message. MqttClientId = {}, messageId = {}.", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
final String topic = evt.getTopic();
final Topic topic = new Topic(evt.getTopic());
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
if (LOG.isDebugEnabled()) {
broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java
+2 -1
View File
@@ -4,6 +4,7 @@ package io.moquette.spi.impl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.Channel;
@@ -17,7 +18,7 @@ abstract class QosPublishHandler {
this.m_authorizator = m_authorizator;
}
public boolean checkWriteOnTopic(String topic, Channel channel) {
public boolean checkWriteOnTopic(Topic topic, Channel channel) {
String clientID = NettyUtils.clientID(channel);
String username = NettyUtils.userName(channel);
if (!m_authorizator.canWrite(topic, username, clientID)) {
broker/src/main/java/io/moquette/spi/impl/security/Authorization.java
+4 -3
View File
@@ -17,6 +17,7 @@
package io.moquette.spi.impl.security;
import static io.moquette.spi.impl.security.Authorization.Permission.READWRITE;
import io.moquette.spi.impl.subscriptions.Topic;
/**
* Carries the read/write authorization to topics for the users.
@@ -25,7 +26,7 @@ import static io.moquette.spi.impl.security.Authorization.Permission.READWRITE;
*/
public class Authorization {
protected final String topic;
protected final Topic topic;
protected final Permission permission;
/**
@@ -35,11 +36,11 @@ public class Authorization {
READ, WRITE, READWRITE
}
Authorization(String topic) {
Authorization(Topic topic) {
this(topic, Permission.READWRITE);
}
Authorization(String topic, Permission permission) {
Authorization(Topic topic, Permission permission) {
this.topic = topic;
this.permission = permission;
}
broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java
+10 -10
View File
@@ -16,7 +16,7 @@
package io.moquette.spi.impl.security;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -98,37 +98,37 @@ class AuthorizationsCollector implements IAuthorizator {
try {
Authorization.Permission permission = Authorization.Permission.valueOf(tokens[1].toUpperCase());
// bring topic with all original spacing
String topic = line.substring(line.indexOf(tokens[2]));
Topic topic = new Topic(line.substring(line.indexOf(tokens[2])));
return new Authorization(topic, permission);
} catch (IllegalArgumentException iaex) {
throw new ParseException("invalid permission token", 1);
}
}
String topic = tokens[1];
Topic topic = new Topic(tokens[1]);
return new Authorization(topic);
}
@Override
public boolean canWrite(String topic, String user, String client) {
public boolean canWrite(Topic topic, String user, String client) {
return canDoOperation(topic, Authorization.Permission.WRITE, user, client);
}
@Override
public boolean canRead(String topic, String user, String client) {
public boolean canRead(Topic topic, String user, String client) {
return canDoOperation(topic, Authorization.Permission.READ, user, client);
}
private boolean canDoOperation(String topic, Authorization.Permission permission, String username, String client) {
private boolean canDoOperation(Topic topic, Authorization.Permission permission, String username, String client) {
if (matchACL(m_globalAuthorizations, topic, permission)) {
return true;
}
if (isNotEmpty(client) || isNotEmpty(username)) {
for (Authorization auth : m_patternAuthorizations) {
String substitutedTopic = auth.topic.replace("%c", client).replace("%u", username);
Topic substitutedTopic = new Topic(auth.topic.toString().replace("%c", client).replace("%u", username));
if (auth.grant(permission)) {
if (SubscriptionsStore.matchTopics(topic, substitutedTopic)) {
if (topic.match(substitutedTopic)) {
return true;
}
}
@@ -146,10 +146,10 @@ class AuthorizationsCollector implements IAuthorizator {
return false;
}
private boolean matchACL(List<Authorization> auths, String topic, Authorization.Permission permission) {
private boolean matchACL(List<Authorization> auths, Topic topic, Authorization.Permission permission) {
for (Authorization auth : auths) {
if (auth.grant(permission)) {
if (SubscriptionsStore.matchTopics(topic, auth.topic)) {
if (topic.match(auth.topic)) {
return true;
}
}
broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java
+3 -2
View File
@@ -16,6 +16,7 @@
package io.moquette.spi.impl.security;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
/**
@@ -24,12 +25,12 @@ import io.moquette.spi.security.IAuthorizator;
public class DenyAllAuthorizator implements IAuthorizator {
@Override
public boolean canWrite(String topic, String user, String client) {
public boolean canWrite(Topic topic, String user, String client) {
return true;
}
@Override
public boolean canRead(String topic, String user, String client) {
public boolean canRead(Topic topic, String user, String client) {
return true;
}
}
broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java
+3 -2
View File
@@ -16,6 +16,7 @@
package io.moquette.spi.impl.security;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
/**
@@ -24,12 +25,12 @@ import io.moquette.spi.security.IAuthorizator;
public class PermitAllAuthorizator implements IAuthorizator {
@Override
public boolean canWrite(String topic, String user, String client) {
public boolean canWrite(Topic topic, String user, String client) {
return true;
}
@Override
public boolean canRead(String topic, String user, String client) {
public boolean canRead(Topic topic, String user, String client) {
return true;
}
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java
+3 -3
View File
@@ -31,10 +31,10 @@ public final class Subscription implements Serializable {
private static final long serialVersionUID = -3383457629635732794L;
final MqttQoS requestedQos; // max QoS acceptable
final String clientId;
final String topicFilter;
final Topic topicFilter;
final boolean active;
public Subscription(String clientId, String topicFilter, MqttQoS requestedQos) {
public Subscription(String clientId, Topic topicFilter, MqttQoS requestedQos) {
this.requestedQos = requestedQos;
this.clientId = clientId;
this.topicFilter = topicFilter;
@@ -56,7 +56,7 @@ public final class Subscription implements Serializable {
return requestedQos;
}
public String getTopicFilter() {
public Topic getTopicFilter() {
return topicFilter;
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+5 -150
View File
@@ -20,7 +20,6 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicReference;
@@ -43,27 +42,6 @@ public class SubscriptionsStore {
}
}
/**
* Check if the topic filter of the subscription is well formed
*
* @param topicFilter
* the filter to validate
* @return true if it's correct.
*/
public static boolean validate(String topicFilter) {
try {
parseTopic(topicFilter);
return true;
} catch (ParseException pex) {
LOG.warn(
"The topic filter is malformed. TopicFilter = {}, cause = {}, errorMessage = {}.",
topicFilter,
pex.getCause(),
pex.getMessage());
return false;
}
}
public interface IVisitor<T> {
void visit(TreeNode node, int deep);
@@ -150,24 +128,11 @@ public class SubscriptionsStore {
LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot);
}
protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) {
List<Token> tokens;
try {
tokens = parseTopic(topic);
} catch (ParseException ex) {
// TODO handle the parse exception
LOG.error(
"The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.",
topic,
ex.getCause(),
ex.getMessage());
throw new IllegalArgumentException(ex.getMessage());
}
protected NodeCouple recreatePath(Topic topic, final TreeNode oldRoot) {
final TreeNode newRoot = oldRoot.copy();
TreeNode parent = newRoot;
TreeNode current = newRoot;
for (Token token : tokens) {
for (Token token : topic.getTokens()) {
TreeNode matchingChildren;
// check if a children with the same token already exists
@@ -188,7 +153,7 @@ public class SubscriptionsStore {
return new NodeCouple(newRoot, current);
}
public void removeSubscription(String topic, String clientID) {
public void removeSubscription(Topic topic, String clientID) {
/*
* The topic filters have already been validated at the ProtocolProcessor. We can assume
* they are valid.
@@ -230,21 +195,8 @@ public class SubscriptionsStore {
* to use fo searching matching subscriptions.
* @return the list of matching subscriptions, or empty if not matching.
*/
public List<Subscription> matches(String topic) {
List<Token> tokens;
try {
tokens = parseTopic(topic);
} catch (ParseException ex) {
// TODO handle the parse exception
LOG.warn(
"The topic is malformed. Topic = {}, cause = {}, errorMessage = {}.",
topic,
ex.getCause(),
ex.getMessage());
return Collections.emptyList();
}
Queue<Token> tokenQueue = new LinkedBlockingDeque<>(tokens);
public List<Subscription> matches(Topic topic) {
Queue<Token> tokenQueue = new LinkedBlockingDeque<>(topic.getTokens());
List<ClientTopicCouple> matchingSubs = new ArrayList<>();
subscriptions.get().matches(tokenQueue, matchingSubs);
@@ -288,101 +240,4 @@ public class SubscriptionsStore {
bfsVisit(child, visitor, ++deep);
}
}
/**
* 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
* @return true if the two topics match.
*/
// TODO reimplement with iterators or with queues
public static boolean matchTopics(String msgTopic, String subscriptionTopic) {
try {
List<Token> msgTokens = SubscriptionsStore.parseTopic(msgTopic);
List<Token> subscriptionTokens = SubscriptionsStore.parseTopic(subscriptionTopic);
int i = 0;
for (; i < subscriptionTokens.size(); i++) {
Token subToken = subscriptionTokens.get(i);
if (subToken != Token.MULTI && subToken != Token.SINGLE) {
if (i >= msgTokens.size()) {
return false;
}
Token msgToken = msgTokens.get(i);
if (!msgToken.equals(subToken)) {
return false;
}
} else {
if (subToken == Token.MULTI) {
return true;
}
if (subToken == Token.SINGLE) {
// skip a step forward
}
}
}
// if last token was a SINGLE then treat it as an empty
// if (subToken == Token.SINGLE && (i - msgTokens.size() == 1)) {
// i--;
// }
return i == msgTokens.size();
} catch (ParseException ex) {
LOG.error(
"The message topic, the subscription topic or both are malformed. "
+ "MsgTopic = {}, subscriptionTopic = {}, cause = {}, errorMessage = {}.",
msgTopic,
subscriptionTopic,
ex.getCause(),
ex.getMessage());
throw new IllegalStateException(ex.getMessage());
}
}
protected static List<Token> parseTopic(String topic) throws ParseException {
List<Token> res = new ArrayList<>();
String[] splitted = topic.split("/");
if (splitted.length == 0) {
res.add(Token.EMPTY);
}
if (topic.endsWith("/")) {
// Add a fictious space
String[] newSplitted = new String[splitted.length + 1];
System.arraycopy(splitted, 0, newSplitted, 0, splitted.length);
newSplitted[splitted.length] = "";
splitted = newSplitted;
}
for (int i = 0; i < splitted.length; i++) {
String s = splitted[i];
if (s.isEmpty()) {
// if (i != 0) {
// throw new ParseException("Bad format of topic, expetec topic name between
// separators", i);
// }
res.add(Token.EMPTY);
} else if (s.equals("#")) {
// check that multi is the last symbol
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);
}
res.add(Token.MULTI);
} else if (s.contains("#")) {
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
} else if (s.equals("+")) {
res.add(Token.SINGLE);
} else if (s.contains("+")) {
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
} else {
res.add(new Token(s));
}
}
return res;
}
}
broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java
+172
View File
@@ -0,0 +1,172 @@
/*
* Copyright (c) 2012-2017 The original author or authorsgetRockQuestions()
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi.impl.subscriptions;
import java.io.Serializable;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class Topic implements Serializable {
private static final Logger LOG = LoggerFactory.getLogger(Topic.class);
private static final long serialVersionUID = 2438799283749822L;
private final String topic;
private transient List<Token> tokens;
private transient boolean valid;
public Topic(String topic) {
this.topic = topic;
}
@Override
public String toString() {
return topic;
}
public List<Token> getTokens() {
if (tokens == null) {
try {
tokens = parseTopic(topic);
valid = true;
} catch (ParseException e) {
valid = false;
LOG.error(e.toString(), e);
}
}
return tokens;
}
private List<Token> parseTopic(String topic) throws ParseException {
List<Token> res = new ArrayList<>();
String[] splitted = topic.split("/");
if (splitted.length == 0) {
res.add(Token.EMPTY);
}
if (topic.endsWith("/")) {
// Add a fictious space
String[] newSplitted = new String[splitted.length + 1];
System.arraycopy(splitted, 0, newSplitted, 0, splitted.length);
newSplitted[splitted.length] = "";
splitted = newSplitted;
}
for (int i = 0; i < splitted.length; i++) {
String s = splitted[i];
if (s.isEmpty()) {
// if (i != 0) {
// throw new ParseException("Bad format of topic, expetec topic name between
// separators", i);
// }
res.add(Token.EMPTY);
} else if (s.equals("#")) {
// check that multi is the last symbol
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);
}
res.add(Token.MULTI);
} else if (s.contains("#")) {
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
} else if (s.equals("+")) {
res.add(Token.SINGLE);
} else if (s.contains("+")) {
throw new ParseException("Bad format of topic, invalid subtopic name: " + s, i);
} else {
res.add(new Token(s));
}
}
return res;
}
public boolean isValid() {
if (tokens == null)
getTokens();
return valid;
}
/**
* 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
* @return true if the two topics match.
*/
// TODO reimplement with iterators or with queues
public boolean match(Topic subscriptionTopic) {
List<Token> msgTokens = getTokens();
List<Token> subscriptionTokens = subscriptionTopic.getTokens();
int i = 0;
for (; i < subscriptionTokens.size(); i++) {
Token subToken = subscriptionTokens.get(i);
if (subToken != Token.MULTI && subToken != Token.SINGLE) {
if (i >= msgTokens.size()) {
return false;
}
Token msgToken = msgTokens.get(i);
if (!msgToken.equals(subToken)) {
return false;
}
} else {
if (subToken == Token.MULTI) {
return true;
}
if (subToken == Token.SINGLE) {
// skip a step forward
}
}
}
// if last token was a SINGLE then treat it as an empty
// if (subToken == Token.SINGLE && (i - msgTokens.size() == 1)) {
// i--;
// }
return i == msgTokens.size();
}
@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
if (getClass() != obj.getClass()) {
return false;
}
Topic other = (Topic) obj;
return Objects.equals(this.topic, other.topic);
}
@Override
public int hashCode() {
return topic.hashCode();
}
}
broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+5 -4
View File
@@ -19,6 +19,7 @@ package io.moquette.spi.persistence;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Topic;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +38,7 @@ class MapDBMessagesStore implements IMessagesStore {
private DB m_db;
// maps clientID -> guid
private ConcurrentMap<String, MessageGUID> m_retainedStore;
private ConcurrentMap<Topic, MessageGUID> m_retainedStore;
// maps guid to message, it's message store
private ConcurrentMap<MessageGUID, IMessagesStore.StoredMessage> m_persistentMessageStore;
@@ -53,7 +54,7 @@ class MapDBMessagesStore implements IMessagesStore {
}
@Override
public void storeRetained(String topic, MessageGUID guid) {
public void storeRetained(Topic topic, MessageGUID guid) {
LOG.debug("Storing retained messages. Topic = {}, guid = {}.", topic, guid);
m_retainedStore.put(topic, guid);
}
@@ -62,7 +63,7 @@ class MapDBMessagesStore implements IMessagesStore {
public Collection<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("Scanning retained messages...");
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, MessageGUID> entry : m_retainedStore.entrySet()) {
for (Map.Entry<Topic, MessageGUID> entry : m_retainedStore.entrySet()) {
final MessageGUID guid = entry.getValue();
StoredMessage storedMsg = m_persistentMessageStore.get(guid);
if (condition.match(entry.getKey())) {
@@ -127,7 +128,7 @@ class MapDBMessagesStore implements IMessagesStore {
}
@Override
public void cleanRetained(String topic) {
public void cleanRetained(Topic topic) {
LOG.debug("Cleaning retained messages. Topic = {}.", topic);
m_retainedStore.remove(topic);
}
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+6 -5
View File
@@ -22,6 +22,7 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.Utils;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
import org.mapdb.DB;
import org.slf4j.Logger;
@@ -83,7 +84,7 @@ class MapDBSessionsStore implements ISessionsStore {
}
@Override
public void removeSubscription(String topicFilter, String clientID) {
public void removeSubscription(Topic topicFilter, String clientID) {
LOG.info("Removing subscription. CId= {}, topics = {}.", clientID, topicFilter);
if (!m_db.exists("subscriptions_" + clientID)) {
return;
@@ -115,8 +116,8 @@ class MapDBSessionsStore implements ISessionsStore {
LOG.info("Retrieving existing subscriptions...");
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()) {
ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
for (Topic topicFilter : clientSubscriptions.keySet()) {
allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter));
}
}
@@ -128,7 +129,7 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public Subscription getSubscription(ClientTopicCouple couple) {
ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID);
LOG.info("Retrieving subscriptions. CId= {}, subscriptions = {}.", couple.clientID, clientSubscriptions);
return clientSubscriptions.get(couple.topicFilter);
}
@@ -138,7 +139,7 @@ class MapDBSessionsStore implements ISessionsStore {
LOG.info("Retrieving existing subscriptions...");
List<Subscription> subscriptions = new ArrayList<>();
for (String clientID : m_persistentSessions.keySet()) {
ConcurrentMap<String, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
ConcurrentMap<Topic, Subscription> clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID);
subscriptions.addAll(clientSubscriptions.values());
}
LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions);
broker/src/main/java/io/moquette/spi/security/IAuthorizator.java
+4 -2
View File
@@ -16,6 +16,8 @@
package io.moquette.spi.security;
import io.moquette.spi.impl.subscriptions.Topic;
/**
* ACL checker.
*
@@ -37,7 +39,7 @@ public interface IAuthorizator {
* the client
* @return true if the user from client can publish data on topic.
*/
boolean canWrite(String topic, String user, String client);
boolean canWrite(Topic topic, String user, String client);
boolean canRead(String topic, String user, String client);
boolean canRead(Topic topic, String user, String client);
}
broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java
+3 -3
View File
@@ -19,11 +19,11 @@ package io.moquette.server;
import io.moquette.BrokerConstants;
import io.moquette.server.config.IConfig;
import io.moquette.server.config.MemoryConfig;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthenticator;
import io.moquette.spi.security.IAuthorizator;
import org.junit.After;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.util.Properties;
import static org.junit.Assert.*;
@@ -70,12 +70,12 @@ public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizat
}
@Override
public boolean canWrite(String topic, String user, String client) {
public boolean canWrite(Topic topic, String user, String client) {
return true;
}
@Override
public boolean canRead(String topic, String user, String client) {
public boolean canRead(Topic topic, String user, String client) {
return true;
}
broker/src/test/java/io/moquette/spi/ClientSessionTest.java
+4 -3
View File
@@ -4,6 +4,7 @@ package io.moquette.spi;
import io.moquette.spi.impl.MemoryStorageService;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import static io.netty.handler.codec.mqtt.MqttQoS.*;
import org.junit.Before;
import org.junit.Test;
@@ -32,17 +33,17 @@ public class ClientSessionTest {
@Test
public void overridingSubscriptions() {
// Subscribe on /topic with QOSType.MOST_ONE
Subscription oldSubscription = new Subscription(session1.clientID, "/topic", AT_MOST_ONCE);
Subscription oldSubscription = new Subscription(session1.clientID, new Topic("/topic"), AT_MOST_ONCE);
session1.subscribe(oldSubscription);
store.add(oldSubscription.asClientTopicCouple());
// Subscribe on /topic again that overrides the previous subscription.
Subscription overrindingSubscription = new Subscription(session1.clientID, "/topic", EXACTLY_ONCE);
Subscription overrindingSubscription = new Subscription(session1.clientID, new Topic("/topic"), EXACTLY_ONCE);
session1.subscribe(overrindingSubscription);
store.add(overrindingSubscription.asClientTopicCouple());
// Verify
List<Subscription> subscriptions = store.matches("/topic");
List<Subscription> subscriptions = store.matches(new Topic("/topic"));
assertEquals(1, subscriptions.size());
Subscription sub = subscriptions.get(0);
assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos());
broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java
+3 -2
View File
@@ -20,6 +20,7 @@ import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.*;
import io.moquette.server.netty.MessageBuilder;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -133,7 +134,7 @@ public class BrokerInterceptorTest {
@Test
public void testNotifyTopicSubscribed() throws Exception {
interceptor.notifyTopicSubscribed(new Subscription("cli1", "o2", MqttQoS.AT_MOST_ONCE), "cli1234");
interceptor.notifyTopicSubscribed(new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE), "cli1234");
interval();
assertEquals(70, n.get());
}
@@ -153,7 +154,7 @@ public class BrokerInterceptorTest {
interceptor.addInterceptHandler(interceptHandlerMock1);
interceptor.addInterceptHandler(interceptHandlerMock2);
Subscription subscription = new Subscription("cli1", "o2", MqttQoS.AT_MOST_ONCE);
Subscription subscription = new Subscription("cli1", new Topic("o2"), MqttQoS.AT_MOST_ONCE);
interceptor.notifyTopicSubscribed(subscription, "cli1234");
interval();
broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java
+5 -4
View File
@@ -19,6 +19,7 @@ package io.moquette.spi.impl;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
@@ -31,7 +32,7 @@ public class MemoryMessagesStore implements IMessagesStore {
private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class);
private Map<String, MessageGUID> m_retainedStore = new HashMap<>();
private Map<Topic, MessageGUID> m_retainedStore = new HashMap<>();
private Map<MessageGUID, StoredMessage> m_persistentMessageStore = new HashMap<>();
private Map<String, Map<Integer, MessageGUID>> m_messageToGuids;
@@ -44,7 +45,7 @@ public class MemoryMessagesStore implements IMessagesStore {
}
@Override
public void storeRetained(String topic, MessageGUID guid) {
public void storeRetained(Topic topic, MessageGUID guid) {
m_retainedStore.put(topic, guid);
}
@@ -54,7 +55,7 @@ public class MemoryMessagesStore implements IMessagesStore {
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, MessageGUID> entry : m_retainedStore.entrySet()) {
for (Map.Entry<Topic, MessageGUID> entry : m_retainedStore.entrySet()) {
final MessageGUID guid = entry.getValue();
StoredMessage storedMsg = m_persistentMessageStore.get(guid);
if (condition.match(entry.getKey())) {
@@ -96,7 +97,7 @@ public class MemoryMessagesStore implements IMessagesStore {
}
@Override
public void cleanRetained(String topic) {
public void cleanRetained(Topic topic) {
m_retainedStore.remove(topic);
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+38 -22
View File
@@ -26,6 +26,7 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.security.PermitAllAuthorizator;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.security.IAuthorizator;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.*;
@@ -110,14 +111,14 @@ public class ProtocolProcessorTest {
@Test
public void testPublishToItself() throws InterruptedException {
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE);
// subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() {
@Override
public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) {
public List<Subscription> matches(Topic topic) {
if (topic.toString().equals(FAKE_TOPIC)) {
return Collections.singletonList(subscription);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
@@ -155,15 +156,18 @@ public class ProtocolProcessorTest {
@Test
public void testPublishToMultipleSubscribers() throws InterruptedException {
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
final Subscription subscriptionClient2 = new Subscription(FAKE_CLIENT_ID2, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE);
final Subscription subscriptionClient2 = new Subscription(
FAKE_CLIENT_ID2,
new Topic(FAKE_TOPIC),
MqttQoS.AT_MOST_ONCE);
// subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() {
@Override
public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) {
public List<Subscription> matches(Topic topic) {
if (topic.toString().equals(FAKE_TOPIC)) {
return Arrays.asList(subscription, subscriptionClient2);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
@@ -227,7 +231,10 @@ public class ProtocolProcessorTest {
// Verify
assertTrue(m_channel.readOutbound() instanceof MqttSubAckMessage);
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
Subscription expectedSubscription = new Subscription(
FAKE_CLIENT_ID,
new Topic(FAKE_TOPIC),
MqttQoS.AT_MOST_ONCE);
assertTrue(subscriptions.contains(expectedSubscription));
}
@@ -237,7 +244,9 @@ public class ProtocolProcessorTest {
NettyUtils.userName(m_channel, fakeUserName);
IAuthorizator mockAuthorizator = mock(IAuthorizator.class);
when(mockAuthorizator.canRead(eq(FAKE_TOPIC), eq(fakeUserName), eq(FAKE_CLIENT_ID))).thenReturn(false);
when(mockAuthorizator.canRead(eq(new Topic(FAKE_TOPIC)), eq(fakeUserName), eq(FAKE_CLIENT_ID)))
.thenReturn(false);
m_processor.init(
subscriptions,
m_messagesStore,
@@ -281,7 +290,11 @@ public class ProtocolProcessorTest {
// Verify
assertEquals(1, subscriptions.size());
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
Subscription expectedSubscription = new Subscription(
FAKE_CLIENT_ID,
new Topic(FAKE_TOPIC),
MqttQoS.AT_MOST_ONCE);
assertTrue(subscriptions.contains(expectedSubscription));
}
@@ -321,14 +334,17 @@ public class ProtocolProcessorTest {
@Test
public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
// simulate a connect that register a clientID to an IoSession
final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
final Subscription subscription = new Subscription(
FAKE_PUBLISHER_ID,
new Topic(FAKE_TOPIC),
MqttQoS.AT_MOST_ONCE);
// subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() {
@Override
public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) {
public List<Subscription> matches(Topic topic) {
if (topic.toString().equals(FAKE_TOPIC)) {
return Collections.singletonList(subscription);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
@@ -377,7 +393,7 @@ public class ProtocolProcessorTest {
public void testRepublishAndConsumePersistedMessages_onReconnect() {
SubscriptionsStore subs = mock(SubscriptionsStore.class);
List<Subscription> emptySubs = Collections.emptyList();
when(subs.matches(anyString())).thenReturn(emptySubs);
when(subs.matches(any(Topic.class))).thenReturn(emptySubs);
StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(), MqttQoS.EXACTLY_ONCE, "/topic");
retainedMessage.setRetained(true);
@@ -410,9 +426,9 @@ public class ProtocolProcessorTest {
m_sessionStore.createNewSession("Subscriber", false);
SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class);
Subscription inactiveSub = new Subscription("Subscriber", "/topic", MqttQoS.AT_LEAST_ONCE);
Subscription inactiveSub = new Subscription("Subscriber", new Topic("/topic"), MqttQoS.AT_LEAST_ONCE);
List<Subscription> inactiveSubscriptions = Collections.singletonList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
when(mockedSubscriptions.matches(eq(new Topic("/topic")))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor();
m_processor.init(
mockedSubscriptions,
@@ -452,8 +468,8 @@ public class ProtocolProcessorTest {
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, FAKE_TOPIC);
public boolean match(Topic key) {
return key.match(new Topic(FAKE_TOPIC));
}
});
assertFalse(messages.isEmpty());
@@ -467,8 +483,8 @@ public class ProtocolProcessorTest {
// Verify
messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, FAKE_TOPIC);
public boolean match(Topic key) {
return key.match(new Topic(FAKE_TOPIC));
}
});
assertTrue(messages.isEmpty());
@@ -476,10 +492,10 @@ public class ProtocolProcessorTest {
@Test
public void testLowerTheQosToTheRequestedBySubscription() {
Subscription subQos1 = new Subscription("Sub A", "a/b", MqttQoS.AT_LEAST_ONCE);
Subscription subQos1 = new Subscription("Sub A", new Topic("a/b"), MqttQoS.AT_LEAST_ONCE);
assertEquals(MqttQoS.AT_LEAST_ONCE, lowerQosToTheSubscriptionDesired(subQos1, MqttQoS.EXACTLY_ONCE));
Subscription subQos2 = new Subscription("Sub B", "a/+", MqttQoS.EXACTLY_ONCE);
Subscription subQos2 = new Subscription("Sub B", new Topic("a/+"), MqttQoS.EXACTLY_ONCE);
assertEquals(MqttQoS.EXACTLY_ONCE, lowerQosToTheSubscriptionDesired(subQos2, MqttQoS.EXACTLY_ONCE));
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+2 -1
View File
@@ -24,6 +24,7 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.impl.security.PermitAllAuthorizator;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
@@ -330,7 +331,7 @@ public class ProtocolProcessor_CONNECT_Test {
m_processor.processSubscribe(m_session, subscribeMsg);
Subscription expectedSubscription = new Subscription(
ProtocolProcessorTest.FAKE_CLIENT_ID,
ProtocolProcessorTest.FAKE_TOPIC,
new Topic(ProtocolProcessorTest.FAKE_TOPIC),
MqttQoS.AT_MOST_ONCE);
assertTrue(subscriptions.contains(expectedSubscription));
assertEqualsSubAck(m_session.readOutbound());
broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java
+3 -2
View File
@@ -17,6 +17,7 @@
package io.moquette.spi.impl.security;
import org.junit.Test;
import io.moquette.spi.impl.subscriptions.Topic;
import java.io.Reader;
import java.io.StringReader;
import java.text.ParseException;
@@ -57,8 +58,8 @@ public class ACLFileParserTest {
AuthorizationsCollector authorizations = ACLFileParser.parse(conf);
// Verify
assertTrue(authorizations.canRead("/weather/italy/anemometer", "", ""));
assertTrue(authorizations.canWrite("/weather/italy/anemometer", "", ""));
assertTrue(authorizations.canRead(new Topic("/weather/italy/anemometer"), "", ""));
assertTrue(authorizations.canWrite(new Topic("/weather/italy/anemometer"), "", ""));
}
}
broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java
+15 -14
View File
@@ -18,6 +18,7 @@ package io.moquette.spi.impl.security;
import org.junit.Before;
import org.junit.Test;
import io.moquette.spi.impl.subscriptions.Topic;
import java.text.ParseException;
import static org.junit.Assert.*;
@@ -26,12 +27,12 @@ import static org.junit.Assert.*;
*/
public class AuthorizationsCollectorTest {
private static final Authorization RW_ANEMOMETER = new Authorization("/weather/italy/anemometer");
private static final Authorization RW_ANEMOMETER = new Authorization(new Topic("/weather/italy/anemometer"));
private static final Authorization R_ANEMOMETER = new Authorization(
"/weather/italy/anemometer",
new Topic("/weather/italy/anemometer"),
Authorization.Permission.READ);
private static final Authorization W_ANEMOMETER = new Authorization(
"/weather/italy/anemometer",
new Topic("/weather/italy/anemometer"),
Authorization.Permission.WRITE);
private AuthorizationsCollector authorizator;
@@ -75,7 +76,7 @@ public class AuthorizationsCollectorTest {
@Test
public void testParseAuthLineValid_topic_with_space() throws ParseException {
Authorization expected = new Authorization("/weather/eastern italy/anemometer");
Authorization expected = new Authorization(new Topic("/weather/eastern italy/anemometer"));
Authorization authorization = authorizator.parseAuthLine("topic readwrite /weather/eastern italy/anemometer");
// Verify
@@ -92,7 +93,7 @@ public class AuthorizationsCollectorTest {
authorizator.parse("topic write /sensors");
// verify
assertTrue(authorizator.canWrite("/sensors", "", ""));
assertTrue(authorizator.canWrite(new Topic("/sensors"), "", ""));
}
@Test
@@ -100,7 +101,7 @@ public class AuthorizationsCollectorTest {
authorizator.parse("topic read /sensors");
// verify
assertTrue(authorizator.canRead("/sensors", "", ""));
assertTrue(authorizator.canRead(new Topic("/sensors"), "", ""));
}
@Test
@@ -109,8 +110,8 @@ public class AuthorizationsCollectorTest {
authorizator.parse("topic read /sensors/anemometer");
// verify
assertTrue(authorizator.canWrite("/sensors", "", ""));
assertFalse(authorizator.canRead("/sensors", "", ""));
assertTrue(authorizator.canWrite(new Topic("/sensors"), "", ""));
assertFalse(authorizator.canRead(new Topic("/sensors"), "", ""));
}
@Test
@@ -118,7 +119,7 @@ public class AuthorizationsCollectorTest {
authorizator.parse("topic write /sensors/#");
// verify
assertTrue(authorizator.canWrite("/sensors/anemometer/wind", "", ""));
assertTrue(authorizator.canWrite(new Topic("/sensors/anemometer/wind"), "", ""));
}
@Test
@@ -126,7 +127,7 @@ public class AuthorizationsCollectorTest {
authorizator.parse("topic write /sensors/+");
// verify
assertTrue(authorizator.canWrite("/sensors/anemometer", "", ""));
assertTrue(authorizator.canWrite(new Topic("/sensors/anemometer"), "", ""));
}
@Test
@@ -135,8 +136,8 @@ public class AuthorizationsCollectorTest {
authorizator.parse("topic write /sensors");
// verify
assertTrue(authorizator.canWrite("/sensors", "john", ""));
assertFalse(authorizator.canWrite("/sensors", "jack", ""));
assertTrue(authorizator.canWrite(new Topic("/sensors"), "john", ""));
assertFalse(authorizator.canWrite(new Topic("/sensors"), "jack", ""));
}
@Test
@@ -144,7 +145,7 @@ public class AuthorizationsCollectorTest {
authorizator.parse("pattern read /weather/italy/%c");
// Verify
assertTrue(authorizator.canRead("/weather/italy/anemometer1", "", "anemometer1"));
assertTrue(authorizator.canRead(new Topic("/weather/italy/anemometer1"), "", "anemometer1"));
}
@Test
@@ -152,6 +153,6 @@ public class AuthorizationsCollectorTest {
authorizator.parse("pattern read /weather/%u/%c");
// Verify
assertTrue(authorizator.canRead("/weather/italy/anemometer1", "italy", "anemometer1"));
assertTrue(authorizator.canRead(new Topic("/weather/italy/anemometer1"), "italy", "anemometer1"));
}
}
broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java
+67 -172
View File
@@ -24,7 +24,6 @@ import io.netty.handler.codec.mqtt.MqttQoS;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.text.ParseException;
import java.util.List;
import static org.assertj.core.api.Assertions.*;
@@ -49,179 +48,136 @@ public class SubscriptionsStoreTest {
store.init(sessionsStore);
}
@Test
public void testParseTopic() throws ParseException {
List<Token> tokens = SubscriptionsStore.parseTopic("finance/stock/ibm");
assertThat(tokens).containsExactlyInAnyOrder(asArray("finance", "stock", "ibm"));
tokens = SubscriptionsStore.parseTopic("/finance/stock/ibm");
assertThat(tokens).containsExactlyInAnyOrder(asArray(Token.EMPTY, "finance", "stock", "ibm"));
tokens = SubscriptionsStore.parseTopic("/");
assertThat(tokens).containsExactlyInAnyOrder(asArray(Token.EMPTY, Token.EMPTY));
}
@Test
public void testParseTopicMultiValid() throws ParseException {
List<Token> tokens = SubscriptionsStore.parseTopic("finance/stock/#");
assertThat(tokens).containsExactlyInAnyOrder(asArray("finance", "stock", Token.MULTI));
tokens = SubscriptionsStore.parseTopic("#");
assertThat(tokens).containsExactlyInAnyOrder(asArray(Token.MULTI));
}
@Test(expected = ParseException.class)
public void testParseTopicMultiInTheMiddleNotValid() throws ParseException {
SubscriptionsStore.parseTopic("finance/#/closingprice");
}
@Test(expected = ParseException.class)
public void testParseTopicMultiNotAfterSeparatorNotValid() throws ParseException {
SubscriptionsStore.parseTopic("finance#");
}
@Test(expected = ParseException.class)
public void testParseTopicMultiNotAlone() throws ParseException {
SubscriptionsStore.parseTopic("/finance/#closingprice");
}
@Test
public void testParseTopicSingleValid() throws ParseException {
List<Token> tokens = SubscriptionsStore.parseTopic("finance/stock/+");
assertThat(tokens).containsExactlyInAnyOrder(asArray("finance", "stock", Token.SINGLE));
tokens = SubscriptionsStore.parseTopic("+");
assertThat(tokens).containsExactlyInAnyOrder(asArray(Token.SINGLE));
tokens = SubscriptionsStore.parseTopic("finance/+/ibm");
assertThat(tokens).containsExactlyInAnyOrder(asArray("finance", Token.SINGLE, "ibm"));
}
@Test(expected = ParseException.class)
public void testParseTopicSingleNotAferSeparatorNotValid() throws ParseException {
SubscriptionsStore.parseTopic("finance+");
}
@Test
public void testMatchSimple() {
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", MqttQoS.AT_MOST_ONCE);
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", new Topic("/"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(slashSub);
store.add(slashSub.asClientTopicCouple());
assertThat(store.matches("finance")).isEmpty();
assertThat(store.matches(new Topic("finance"))).isEmpty();
Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", MqttQoS.AT_MOST_ONCE);
Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", new Topic("/finance"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(slashFinanceSub);
store.add(slashFinanceSub.asClientTopicCouple());
assertThat(store.matches("finance")).isEmpty();
assertThat(store.matches(new Topic("finance"))).isEmpty();
assertThat(store.matches("/finance")).contains(slashFinanceSub);
assertThat(store.matches("/")).contains(slashSub);
assertThat(store.matches(new Topic("/finance"))).contains(slashFinanceSub);
assertThat(store.matches(new Topic("/"))).contains(slashSub);
}
@Test
public void testMatchSimpleMulti() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", MqttQoS.AT_MOST_ONCE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("#"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(anySub);
store.add(anySub.asClientTopicCouple());
assertThat(store.matches("finance")).contains(anySub);
assertThat(store.matches(new Topic("finance"))).contains(anySub);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", MqttQoS.AT_MOST_ONCE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", new Topic("finance/#"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(financeAnySub);
store.add(financeAnySub.asClientTopicCouple());
assertThat(store.matches("finance")).containsExactlyInAnyOrder(financeAnySub, anySub);
assertThat(store.matches(new Topic("finance"))).containsExactlyInAnyOrder(financeAnySub, anySub);
}
@Test
public void testMatchingDeepMulti_one_layer() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", MqttQoS.AT_MOST_ONCE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", MqttQoS.AT_MOST_ONCE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("#"), MqttQoS.AT_MOST_ONCE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", new Topic("finance/#"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(anySub);
sessionsStore.addNewSubscription(financeAnySub);
store.add(anySub.asClientTopicCouple());
store.add(financeAnySub.asClientTopicCouple());
// Verify
assertThat(store.matches("finance/stock")).containsExactlyInAnyOrder(financeAnySub, anySub);
assertThat(store.matches("finance/stock/ibm")).containsExactlyInAnyOrder(financeAnySub, anySub);
assertThat(store.matches(new Topic("finance/stock"))).containsExactlyInAnyOrder(financeAnySub, anySub);
assertThat(store.matches(new Topic("finance/stock/ibm"))).containsExactlyInAnyOrder(financeAnySub, anySub);
}
@Test
public void testMatchingDeepMulti_two_layer() {
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", MqttQoS.AT_MOST_ONCE);
Subscription financeAnySub = new Subscription(
"FAKE_CLI_ID_1",
new Topic("finance/stock/#"),
MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(financeAnySub);
store.add(financeAnySub.asClientTopicCouple());
// Verify
assertThat(store.matches("finance/stock/ibm")).containsExactly(financeAnySub);
assertThat(store.matches(new Topic("finance/stock/ibm"))).containsExactly(financeAnySub);
}
@Test
public void testMatchSimpleSingle() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", MqttQoS.AT_MOST_ONCE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("+"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(anySub);
store.add(anySub.asClientTopicCouple());
assertThat(store.matches("finance")).containsExactly(anySub);
assertThat(store.matches(new Topic("finance"))).containsExactly(anySub);
Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", MqttQoS.AT_MOST_ONCE);
Subscription financeOne = new Subscription("FAKE_CLI_ID_1", new Topic("finance/+"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(financeOne);
store.add(financeOne.asClientTopicCouple());
assertThat(store.matches("finance/stock")).containsExactly(financeOne);
assertThat(store.matches(new Topic("finance/stock"))).containsExactly(financeOne);
}
@Test
public void testMatchManySingle() {
Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", MqttQoS.AT_MOST_ONCE);
Subscription manySub = new Subscription("FAKE_CLI_ID_1", new Topic("+/+"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(manySub);
store.add(manySub.asClientTopicCouple());
// verify
assertThat(store.matches("/finance")).contains(manySub);
assertThat(store.matches(new Topic("/finance"))).contains(manySub);
}
@Test
public void testMatchSlashSingle() {
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", MqttQoS.AT_MOST_ONCE);
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", new Topic("/+"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(slashPlusSub);
store.add(slashPlusSub.asClientTopicCouple());
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", MqttQoS.AT_MOST_ONCE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", new Topic("+"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(anySub);
store.add(anySub.asClientTopicCouple());
// Verify
assertThat(store.matches("/finance")).containsOnly(slashPlusSub);
assertThat(store.matches("/finance")).doesNotContain(anySub);
assertThat(store.matches(new Topic("/finance"))).containsOnly(slashPlusSub);
assertThat(store.matches(new Topic("/finance"))).doesNotContain(anySub);
}
@Test
public void testMatchManyDeepSingle() {
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", MqttQoS.AT_MOST_ONCE);
Subscription slashPlusSub = new Subscription(
"FAKE_CLI_ID_1",
new Topic("/finance/+/ibm"),
MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(slashPlusSub);
store.add(slashPlusSub.asClientTopicCouple());
Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_2", "/+/stock/+", MqttQoS.AT_MOST_ONCE);
Subscription slashPlusDeepSub = new Subscription(
"FAKE_CLI_ID_2",
new Topic("/+/stock/+"),
MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(slashPlusDeepSub);
store.add(slashPlusDeepSub.asClientTopicCouple());
// Verify
assertThat(store.matches("/finance/stock/ibm")).containsExactlyInAnyOrder(slashPlusSub, slashPlusDeepSub);
assertThat(store.matches(new Topic("/finance/stock/ibm")))
.containsExactlyInAnyOrder(slashPlusSub, slashPlusDeepSub);
}
@Test
public void testMatchSimpleMulti_allTheTree() {
Subscription sub = new Subscription("FAKE_CLI_ID_1", "#", MqttQoS.AT_MOST_ONCE);
Subscription sub = new Subscription("FAKE_CLI_ID_1", new Topic("#"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(sub);
store.add(sub.asClientTopicCouple());
assertThat(store.matches("finance")).isNotEmpty();
assertThat(store.matches("finance/ibm")).isNotEmpty();
assertThat(store.matches(new Topic("finance"))).isNotEmpty();
assertThat(store.matches(new Topic("finance/ibm"))).isNotEmpty();
}
@Test
public void testMatchSimpleMulti_zeroLevel() {
// check MULTI in case of zero level match
Subscription sub = new Subscription("FAKE_CLI_ID_1", "finance/#", MqttQoS.AT_MOST_ONCE);
Subscription sub = new Subscription("FAKE_CLI_ID_1", new Topic("finance/#"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(sub);
store.add(sub.asClientTopicCouple());
assertThat(store.matches("finance")).isNotEmpty();
assertThat(store.matches(new Topic("finance"))).isNotEmpty();
}
@Test
@@ -248,7 +204,9 @@ public class SubscriptionsStoreTest {
assertMatch("foo/bar/+", "foo/bar/");
}
private void assertMatch(String subscription, String topic) {
private void assertMatch(String s, String t) {
Topic subscription = new Topic(s);
Topic topic = new Topic(t);
store = new SubscriptionsStore();
MemoryStorageService memStore = new MemoryStorageService();
memStore.initStore();
@@ -260,7 +218,9 @@ public class SubscriptionsStoreTest {
assertThat(store.matches(topic)).isNotEmpty();
}
private void assertNotMatch(String subscription, String topic) {
private void assertNotMatch(String s, String t) {
Topic subscription = new Topic(s);
Topic topic = new Topic(t);
store = new SubscriptionsStore();
MemoryStorageService memStore = new MemoryStorageService();
memStore.initStore();
@@ -274,7 +234,7 @@ public class SubscriptionsStoreTest {
@Test
public void testRemoveClientSubscriptions_existingClientID() {
String cliendID = "FAKE_CLID_1";
Subscription sub = new Subscription(cliendID, "finance/#", MqttQoS.AT_MOST_ONCE);
Subscription sub = new Subscription(cliendID, new Topic("finance/#"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(sub);
store.add(sub.asClientTopicCouple());
@@ -288,7 +248,7 @@ public class SubscriptionsStoreTest {
@Test
public void testRemoveClientSubscriptions_notexistingClientID() {
String clientID = "FAKE_CLID_1";
Subscription s = new Subscription(clientID, "finance/#", MqttQoS.AT_MOST_ONCE);
Subscription s = new Subscription(clientID, new Topic("finance/#"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(s);
store.add(s.asClientTopicCouple());
@@ -301,66 +261,15 @@ public class SubscriptionsStoreTest {
@Test
public void testOverlappingSubscriptions() {
Subscription genericSub = new Subscription("FAKE_CLI_ID_1", "a/+", MqttQoS.EXACTLY_ONCE);
Subscription genericSub = new Subscription("FAKE_CLI_ID_1", new Topic("a/+"), MqttQoS.EXACTLY_ONCE);
sessionsStore.addNewSubscription(genericSub);
store.add(genericSub.asClientTopicCouple());
Subscription specificSub = new Subscription("FAKE_CLI_ID_1", "a/b", MqttQoS.AT_LEAST_ONCE);
Subscription specificSub = new Subscription("FAKE_CLI_ID_1", new Topic("a/b"), MqttQoS.AT_LEAST_ONCE);
sessionsStore.addNewSubscription(specificSub);
store.add(specificSub.asClientTopicCouple());
// Verify
assertThat(store.matches("a/b").size()).isEqualTo(1);
}
@Test
public void testMatchTopics_simple() {
assertThat(SubscriptionsStore.matchTopics("/", "/")).isTrue();
assertThat(SubscriptionsStore.matchTopics("/finance", "/finance")).isTrue();
}
@Test
public void testMatchTopics_multi() {
assertThat(SubscriptionsStore.matchTopics("finance", "#")).isTrue();
assertThat(SubscriptionsStore.matchTopics("finance", "finance/#")).isTrue();
assertThat(SubscriptionsStore.matchTopics("finance/stock", "finance/#")).isTrue();
assertThat(SubscriptionsStore.matchTopics("finance/stock/ibm", "finance/#")).isTrue();
}
@Test
public void testMatchTopics_single() {
assertThat(SubscriptionsStore.matchTopics("finance", "+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("finance/stock", "finance/+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("finance", "finance/+")).isFalse();
assertThat(SubscriptionsStore.matchTopics("/finance", "/+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("/finance", "+")).isFalse();
assertThat(SubscriptionsStore.matchTopics("/finance", "+/+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("/finance/stock/ibm", "/finance/+/ibm")).isTrue();
assertThat(SubscriptionsStore.matchTopics("/", "+/+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("sport/", "sport/+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("/finance/stock", "+")).isFalse();
}
@Test
public void rogerLightMatchTopics() {
assertThat(SubscriptionsStore.matchTopics("foo/bar", "foo/bar")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo/bar", "foo/+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo/bar/baz", "foo/+/baz")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo/bar/baz", "foo/+/#")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo/bar/baz", "#")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo", "foo/bar")).isFalse();
assertThat(SubscriptionsStore.matchTopics("foo/bar/baz", "foo/+")).isFalse();
assertThat(SubscriptionsStore.matchTopics("foo/bar/bar", "foo/+/baz")).isFalse();
assertThat(SubscriptionsStore.matchTopics("fo2/bar/baz", "foo/+/#")).isFalse();
assertThat(SubscriptionsStore.matchTopics("/foo/bar", "#")).isTrue();
assertThat(SubscriptionsStore.matchTopics("/foo/bar", "/#")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo/bar", "/#")).isFalse();
assertThat(SubscriptionsStore.matchTopics("foo//bar", "foo//bar")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo//bar", "foo//+")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo///baz", "foo/+/+/baz")).isTrue();
assertThat(SubscriptionsStore.matchTopics("foo/bar/", "foo/bar/+")).isTrue();
assertThat(store.matches(new Topic("a/b")).size()).isEqualTo(1);
}
@Test
@@ -372,20 +281,20 @@ public class SubscriptionsStoreTest {
aStore.init(sessionsStore);
// subscribe a not active clientID1 to /topic
sessionsStore.createNewSession("FAKE_CLI_ID_1", true);
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", MqttQoS.AT_MOST_ONCE);
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", new Topic("/topic"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(slashSub);
aStore.add(slashSub.asClientTopicCouple());
// subscribe an active clientID2 to /topic
Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", "/topic", MqttQoS.AT_MOST_ONCE);
Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", new Topic("/topic"), MqttQoS.AT_MOST_ONCE);
sessionsStore.addNewSubscription(slashSub2);
aStore.add(new ClientTopicCouple("FAKE_CLI_ID_2", "/topic"));
aStore.add(new ClientTopicCouple("FAKE_CLI_ID_2", new Topic("/topic")));
// Exercise
aStore.removeSubscription("/topic", slashSub2.getClientId());
aStore.removeSubscription(new Topic("/topic"), slashSub2.getClientId());
// Verify
Subscription remainedSubscription = aStore.matches("/topic").get(0);
Subscription remainedSubscription = aStore.matches(new Topic("/topic")).get(0);
assertThat(remainedSubscription.getClientId()).isEqualTo(slashSub.getClientId());
}
@@ -395,22 +304,22 @@ public class SubscriptionsStoreTest {
@Test
public void duplicatedSubscriptionsWithDifferentQos() {
ClientSession session2 = sessionsStore.createNewSession("client2", true);
Subscription client2Sub = new Subscription("client2", "client/test/b", MqttQoS.AT_MOST_ONCE);
Subscription client2Sub = new Subscription("client2", new Topic("client/test/b"), MqttQoS.AT_MOST_ONCE);
session2.subscribe(client2Sub);
store.add(client2Sub.asClientTopicCouple());
ClientSession session1 = sessionsStore.createNewSession("client1", true);
Subscription client1SubQoS0 = new Subscription("client1", "client/test/b", MqttQoS.AT_MOST_ONCE);
Subscription client1SubQoS0 = new Subscription("client1", new Topic("client/test/b"), MqttQoS.AT_MOST_ONCE);
session1.subscribe(client1SubQoS0);
store.add(client1SubQoS0.asClientTopicCouple());
Subscription client1SubQoS2 = new Subscription("client1", "client/test/b", MqttQoS.EXACTLY_ONCE);
Subscription client1SubQoS2 = new Subscription("client1", new Topic("client/test/b"), MqttQoS.EXACTLY_ONCE);
session1.subscribe(client1SubQoS2);
store.add(client1SubQoS2.asClientTopicCouple());
System.out.println(store.dumpTree());
// Verify
List<Subscription> subscriptions = store.matches("client/test/b");
List<Subscription> subscriptions = store.matches(new Topic("client/test/b"));
assertThat(subscriptions).contains(client1SubQoS2);
assertThat(subscriptions).contains(client2Sub);
@@ -424,7 +333,7 @@ public class SubscriptionsStoreTest {
@Test
public void testRecreatePath_emptyRoot() {
TreeNode oldRoot = new TreeNode();
final SubscriptionsStore.NodeCouple resp = store.recreatePath("/finance", oldRoot);
final SubscriptionsStore.NodeCouple resp = store.recreatePath(new Topic("/finance"), oldRoot);
// Verify
assertThat(resp.root).isNotNull();
@@ -436,8 +345,8 @@ public class SubscriptionsStoreTest {
@Test
public void testRecreatePath_1layer_tree() {
TreeNode oldRoot = new TreeNode();
final SubscriptionsStore.NodeCouple respFinance = store.recreatePath("/finance", oldRoot);
final SubscriptionsStore.NodeCouple respPlus = store.recreatePath("/+", respFinance.root);
final SubscriptionsStore.NodeCouple respFinance = store.recreatePath(new Topic("/finance"), oldRoot);
final SubscriptionsStore.NodeCouple respPlus = store.recreatePath(new Topic("/+"), respFinance.root);
// Verify
assertThat(respPlus.root).isNotNull();
@@ -446,18 +355,4 @@ public class SubscriptionsStoreTest {
assertThat(respPlus.root.m_children.get(0).m_children).contains(respPlus.createdNode);
assertThat(respPlus.root.m_children.get(0).m_children).contains(respFinance.createdNode);
}
private static Token[] asArray(Object... l) {
Token[] tokens = new Token[l.length];
for (int i = 0; i < l.length; i++) {
Object o = l[i];
if (o instanceof Token) {
tokens[i] = (Token) o;
} else {
tokens[i] = new Token(o.toString());
}
}
return tokens;
}
}
broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java
+171
View File
@@ -0,0 +1,171 @@
/*
* Copyright (c) 2012-2017 The original author or authorsgetRockQuestions()
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.moquette.spi.impl.subscriptions;
import org.assertj.core.api.AbstractAssert;
import org.assertj.core.api.Assertions;
import org.junit.Test;
public class TopicTest {
@Test
public void testParseTopic() {
assertThat(new Topic("finance/stock/ibm")).containsToken("finance", "stock", "ibm");
assertThat(new Topic("/finance/stock/ibm")).containsToken(Token.EMPTY, "finance", "stock", "ibm");
assertThat(new Topic("/")).containsToken(Token.EMPTY, Token.EMPTY);
}
@Test
public void testParseTopicMultiValid() {
assertThat(new Topic("finance/stock/#")).containsToken("finance", "stock", Token.MULTI);
assertThat(new Topic("#")).containsToken(Token.MULTI);
}
@Test
public void testValidtionProcess() {
// TopicMultiInTheMiddle
assertThat(new Topic("finance/#/closingprice")).isInValid();
// MultiNotAfterSeparator
assertThat(new Topic("finance#")).isInValid();
// TopicMultiNotAlone
assertThat(new Topic("/finance/#closingprice")).isInValid();
// SingleNotAferSeparator
assertThat(new Topic("finance+")).isInValid();
assertThat(new Topic("finance/+")).isValid();
}
@Test
public void testParseTopicSingleValid() {
assertThat(new Topic("finance/stock/+")).containsToken("finance", "stock", Token.SINGLE);
assertThat(new Topic("+")).containsToken(Token.SINGLE);
assertThat(new Topic("finance/+/ibm")).containsToken("finance", Token.SINGLE, "ibm");
}
@Test
public void testMatchTopics_simple() {
assertThat(new Topic("/")).matches("/");
assertThat(new Topic("/finance")).matches("/finance");
}
@Test
public void testMatchTopics_multi() {
assertThat(new Topic("finance")).matches("#");
assertThat(new Topic("finance")).matches("finance/#");
assertThat(new Topic("finance/stock")).matches("finance/#");
assertThat(new Topic("finance/stock/ibm")).matches("finance/#");
}
@Test
public void testMatchTopics_single() {
assertThat(new Topic("finance")).matches("+");
assertThat(new Topic("finance/stock")).matches("finance/+");
assertThat(new Topic("finance")).doesNotMatch("finance/+");
assertThat(new Topic("/finance")).matches("/+");
assertThat(new Topic("/finance")).doesNotMatch("+");
assertThat(new Topic("/finance")).matches("+/+");
assertThat(new Topic("/finance/stock/ibm")).matches("/finance/+/ibm");
assertThat(new Topic("/")).matches("+/+");
assertThat(new Topic("sport/")).matches("sport/+");
assertThat(new Topic("/finance/stock")).doesNotMatch("+");
}
@Test
public void rogerLightMatchTopics() {
assertThat(new Topic("foo/bar")).matches("foo/bar");
assertThat(new Topic("foo/bar")).matches("foo/+");
assertThat(new Topic("foo/bar/baz")).matches("foo/+/baz");
assertThat(new Topic("foo/bar/baz")).matches("foo/+/#");
assertThat(new Topic("foo/bar/baz")).matches("#");
assertThat(new Topic("foo")).doesNotMatch("foo/bar");
assertThat(new Topic("foo/bar/baz")).doesNotMatch("foo/+");
assertThat(new Topic("foo/bar/bar")).doesNotMatch("foo/+/baz");
assertThat(new Topic("fo2/bar/baz")).doesNotMatch("foo/+/#");
assertThat(new Topic("/foo/bar")).matches("#");
assertThat(new Topic("/foo/bar")).matches("/#");
assertThat(new Topic("foo/bar")).doesNotMatch("/#");
assertThat(new Topic("foo//bar")).matches("foo//bar");
assertThat(new Topic("foo//bar")).matches("foo//+");
assertThat(new Topic("foo///baz")).matches("foo/+/+/baz");
assertThat(new Topic("foo/bar/")).matches("foo/bar/+");
}
public static TopicAssert assertThat(Topic topic) {
return new TopicAssert(topic);
}
static class TopicAssert extends AbstractAssert<TopicAssert, Topic> {
TopicAssert(Topic actual) {
super(actual, TopicAssert.class);
}
public TopicAssert matches(String topic) {
Assertions.assertThat(actual.match(new Topic(topic))).isTrue();
return myself;
}
public TopicAssert doesNotMatch(String topic) {
Assertions.assertThat(actual.match(new Topic(topic))).isFalse();
return myself;
}
public TopicAssert containsToken(Object... tokens) {
Assertions.assertThat(actual.getTokens()).containsExactly(asArray(tokens));
return myself;
}
private Token[] asArray(Object... l) {
Token[] tokens = new Token[l.length];
for (int i = 0; i < l.length; i++) {
Object o = l[i];
if (o instanceof Token) {
tokens[i] = (Token) o;
} else {
tokens[i] = new Token(o.toString());
}
}
return tokens;
}
public TopicAssert isValid() {
Assertions.assertThat(actual.isValid()).isTrue();
return myself;
}
public TopicAssert isInValid() {
Assertions.assertThat(actual.isValid()).isFalse();
return myself;
}
}
}
broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java
+6 -2
View File
@@ -26,6 +26,7 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.junit.After;
import org.junit.Before;
@@ -72,11 +73,14 @@ public class MapDBPersistentStoreTest {
ClientSession session1 = m_sessionsStore.createNewSession("SESSION_ID_1", true);
// Subscribe on /topic with QOSType.MOST_ONE
Subscription oldSubscription = new Subscription(session1.clientID, "/topic", MqttQoS.AT_MOST_ONCE);
Subscription oldSubscription = new Subscription(session1.clientID, new Topic("/topic"), MqttQoS.AT_MOST_ONCE);
session1.subscribe(oldSubscription);
// Subscribe on /topic again that overrides the previous subscription.
Subscription overridingSubscription = new Subscription(session1.clientID, "/topic", MqttQoS.EXACTLY_ONCE);
Subscription overridingSubscription = new Subscription(
session1.clientID,
new Topic("/topic"),
MqttQoS.EXACTLY_ONCE);
session1.subscribe(overridingSubscription);
// Verify
broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java
+9 -8
View File
@@ -24,6 +24,7 @@ import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
import io.moquette.spi.impl.Utils;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.Topic;
import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,7 +40,7 @@ public class MemorySessionStore implements ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
private Map<String, Map<String, Subscription>> m_persistentSubscriptions = new HashMap<>();
private Map<String, Map<Topic, Subscription>> m_persistentSubscriptions = new HashMap<>();
private Map<String, MapDBPersistentStore.PersistentSession> m_persistentSessions = new HashMap<>();
@@ -60,12 +61,12 @@ public class MemorySessionStore implements ISessionsStore {
}
@Override
public void removeSubscription(String topic, String clientID) {
public void removeSubscription(Topic topic, String clientID) {
LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
if (!m_persistentSubscriptions.containsKey(clientID)) {
return;
}
Map<String, Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
Map<Topic, Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
clientSubscriptions.remove(topic);
}
@@ -78,7 +79,7 @@ public class MemorySessionStore implements ISessionsStore {
public void addNewSubscription(Subscription newSubscription) {
final String clientID = newSubscription.getClientId();
if (!m_persistentSubscriptions.containsKey(clientID)) {
m_persistentSubscriptions.put(clientID, new HashMap<String, Subscription>());
m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
}
m_persistentSubscriptions.get(clientID).put(newSubscription.getTopicFilter(), newSubscription);
@@ -102,7 +103,7 @@ public class MemorySessionStore 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 HashMap<String, Subscription>());
m_persistentSubscriptions.put(clientID, new HashMap<Topic, Subscription>());
m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
return new ClientSession(clientID, m_messagesStore, this, cleanSession);
}
@@ -134,7 +135,7 @@ public class MemorySessionStore implements ISessionsStore {
@Override
public List<ClientTopicCouple> listAllSubscriptions() {
List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
for (Map.Entry<String, Map<String, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
for (Subscription sub : entry.getValue().values()) {
allSubscriptions.add(sub.asClientTopicCouple());
}
@@ -144,7 +145,7 @@ public class MemorySessionStore implements ISessionsStore {
@Override
public Subscription getSubscription(ClientTopicCouple couple) {
Map<String, Subscription> subscriptions = m_persistentSubscriptions.get(couple.clientID);
Map<Topic, Subscription> subscriptions = m_persistentSubscriptions.get(couple.clientID);
if (subscriptions == null || subscriptions.isEmpty()) {
return null;
}
@@ -154,7 +155,7 @@ public class MemorySessionStore implements ISessionsStore {
@Override
public List<Subscription> getSubscriptions() {
List<Subscription> subscriptions = new ArrayList<>();
for (Map.Entry<String, Map<String, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
for (Map.Entry<String, Map<Topic, Subscription>> entry : m_persistentSubscriptions.entrySet()) {
subscriptions.addAll(entry.getValue().values());
}
return subscriptions;
perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java
+4 -3
View File
@@ -34,10 +34,11 @@ public class PerformanceTest {
ThreadMXBean b = ManagementFactory.getThreadMXBean();
String[] topics = new String[n];
Topic[] topics = new Topic[n];
for (int i = 0; i < n; i++) {
topics[i] = (i % users) + "/" + ((i / users) % devices) + "/" + ((i / (users * devices)) % values) + "/"
+ ((i / (users * devices * values)) % subs);
topics[i] = new Topic(
(i % users) + "/" + ((i / users) % devices) + "/" + ((i / (users * devices)) % values) + "/"
+ ((i / (users * devices * values)) % subs));
Subscription sub = new Subscription("CLI_ID_" + (i % users), topics[i], MqttQoS.AT_MOST_ONCE);
aSessionsStore.addNewSubscription(sub);