broker/Topic: class added

broker/Topic: String Topics replaced by Topic class
broker/Topic#parseTopic: access set to private
broker/SubscriptionsStoreTest: topic tests moved to own class
broker/TopicTest: containsExactly used
broker/TopicTest: TopicAssert class added
broker/TopicTest$TopicAssert#containsToken: method added
broker/TopicTest$TopicAssert#matches: method added

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;
     }
 

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<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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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;