diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java index b84e3176..71d9bd6f 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java @@ -26,7 +26,7 @@ public class InterceptSubscribeMessage implements InterceptMessage { } public String getTopicFilter() { - return subscription.getTopicFilter(); + return subscription.getTopicFilter().toString(); } public String getUsername() { diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java index 8475bb4c..5d61dde6 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java @@ -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); diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 3b2bf3a6..698d1e62 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -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 subscriptionsToRemove = new HashSet<>(); diff --git a/broker/src/main/java/io/moquette/spi/IMatchingCondition.java b/broker/src/main/java/io/moquette/spi/IMatchingCondition.java index 24a73eb9..997a3be1 100644 --- a/broker/src/main/java/io/moquette/spi/IMatchingCondition.java +++ b/broker/src/main/java/io/moquette/spi/IMatchingCondition.java @@ -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); } diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index e89e073d..63ce2925 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -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); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 3d0cd551..8baa9d4e 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -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 diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 5f020e4b..2b402756 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -26,6 +26,7 @@ import io.moquette.spi.*; 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 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 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()); } }); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java index 55f47a18..26c180d6 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -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; } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 8b0bf075..c0534af2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -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; } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index de76ada6..b42d1b4e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -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 topicMatchingSubscriptions = subscriptions.matches(topic); if (LOG.isDebugEnabled()) { diff --git a/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java index ab1af41a..89e9f024 100644 --- a/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java @@ -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)) { diff --git a/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java b/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java index 9fefd3b0..25e9f6b8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/Authorization.java @@ -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; } diff --git a/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java b/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java index 0686028e..d894befa 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/AuthorizationsCollector.java @@ -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 auths, String topic, Authorization.Permission permission) { + private boolean matchACL(List 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; } } diff --git a/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java b/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java index 614a562d..d81638ec 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/DenyAllAuthorizator.java @@ -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; } } diff --git a/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java b/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java index 80d45666..c484d67b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java +++ b/broker/src/main/java/io/moquette/spi/impl/security/PermitAllAuthorizator.java @@ -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; } } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java index f2370236..37df5c3e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java @@ -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; } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 3353e22e..be2e4aaf 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -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 { 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 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 matches(String topic) { - List 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 tokenQueue = new LinkedBlockingDeque<>(tokens); + public List matches(Topic topic) { + Queue tokenQueue = new LinkedBlockingDeque<>(topic.getTokens()); List 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 msgTokens = SubscriptionsStore.parseTopic(msgTopic); - List 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 parseTopic(String topic) throws ParseException { - List 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; - } } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java new file mode 100644 index 00000000..bc196350 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java @@ -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 tokens; + + private transient boolean valid; + + public Topic(String topic) { + this.topic = topic; + } + + @Override + public String toString() { + return topic; + } + + public List getTokens() { + if (tokens == null) { + try { + tokens = parseTopic(topic); + valid = true; + } catch (ParseException e) { + valid = false; + LOG.error(e.toString(), e); + } + } + + return tokens; + } + + private List parseTopic(String topic) throws ParseException { + List 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 msgTokens = getTokens(); + List 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(); + } +} diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java index 62e6545b..c5283da9 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -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 m_retainedStore; + private ConcurrentMap m_retainedStore; // maps guid to message, it's message store private ConcurrentMap 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 searchMatching(IMatchingCondition condition) { LOG.debug("Scanning retained messages..."); List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { + for (Map.Entry 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); } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index a270da15..3976d157 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -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 allSubscriptions = new ArrayList<>(); for (String clientID : m_persistentSessions.keySet()) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); - for (String topicFilter : clientSubscriptions.keySet()) { + ConcurrentMap 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 clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); + ConcurrentMap 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 subscriptions = new ArrayList<>(); for (String clientID : m_persistentSessions.keySet()) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); + ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); subscriptions.addAll(clientSubscriptions.values()); } LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions); diff --git a/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java b/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java index cba7ab5f..050163bc 100644 --- a/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java +++ b/broker/src/main/java/io/moquette/spi/security/IAuthorizator.java @@ -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); } diff --git a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java index 6eb6ea7f..184c101b 100644 --- a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java +++ b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java @@ -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; } diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index 598728b9..35ad1852 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -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 subscriptions = store.matches("/topic"); + List subscriptions = store.matches(new Topic("/topic")); assertEquals(1, subscriptions.size()); Subscription sub = subscriptions.get(0); assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos()); diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index f1c02002..d09cc135 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -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(); diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java index e48a2a46..0e481866 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java @@ -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 m_retainedStore = new HashMap<>(); + private Map m_retainedStore = new HashMap<>(); private Map m_persistentMessageStore = new HashMap<>(); private Map> 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 results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { + for (Map.Entry 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); } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index bdcb957a..9f5f5dcf 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -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 matches(String topic) { - if (topic.equals(FAKE_TOPIC)) { + public List 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 matches(String topic) { - if (topic.equals(FAKE_TOPIC)) { + public List 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 matches(String topic) { - if (topic.equals(FAKE_TOPIC)) { + public List 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 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 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 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)); } } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index af73bb60..b50788fd 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -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()); diff --git a/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java b/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java index c86bc577..a01190c3 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/ACLFileParserTest.java @@ -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"), "", "")); } } diff --git a/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java index 15b0408c..e2c87220 100644 --- a/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/security/AuthorizationsCollectorTest.java @@ -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")); } } diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 52ab5259..37ddb9f6 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -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 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 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 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 subscriptions = store.matches("client/test/b"); + List 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; - } } diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java new file mode 100644 index 00000000..bfa53500 --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java @@ -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 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; + } + } +} diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java index 9f0f4a92..96a4974b 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -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 diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index c4b803e1..aa8a240e 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -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> m_persistentSubscriptions = new HashMap<>(); + private Map> m_persistentSubscriptions = new HashMap<>(); private Map 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 clientSubscriptions = m_persistentSubscriptions.get(clientID); + Map 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()); + m_persistentSubscriptions.put(clientID, new HashMap()); } 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()); + m_persistentSubscriptions.put(clientID, new HashMap()); 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 listAllSubscriptions() { List allSubscriptions = new ArrayList<>(); - for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + for (Map.Entry> 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 subscriptions = m_persistentSubscriptions.get(couple.clientID); + Map 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 getSubscriptions() { List subscriptions = new ArrayList<>(); - for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { subscriptions.addAll(entry.getValue().values()); } return subscriptions; diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java index 7bc7fa22..33646927 100644 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java @@ -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);