From 562c99b9ed3e57387fbd75e678b0b867eda5c6b5 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Wed, 5 Sep 2012 21:26:53 +0200 Subject: [PATCH] Moved SubscriptionStore and its stuff in separate package --- .../messaging/spi/IStorageService.java | 3 +- .../spi/impl/HawtDBStorageService.java | 1 + .../messaging/spi/impl/SimpleMessaging.java | 45 ++++--------------- .../spi/impl/events/SubscribeEvent.java | 2 +- .../{ => subscriptions}/Subscription.java | 2 +- .../SubscriptionsStore.java | 37 ++++++++++++--- .../spi/impl/DummyStorageService.java | 1 + .../spi/impl/SimpleMessagingTest.java | 27 +---------- .../SubscriptionsStoreTest.java | 35 +++++++++++++-- 9 files changed, 77 insertions(+), 76 deletions(-) rename parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/{ => subscriptions}/Subscription.java (97%) rename parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/{ => subscriptions}/SubscriptionsStore.java (90%) rename parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/{ => subscriptions}/SubscriptionsStoreTest.java (85%) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java index d5726f2f..590cdd40 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -1,7 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi; -import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; -import org.dna.mqtt.moquette.messaging.spi.impl.Subscription; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import static org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService.StoredMessage; diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 6d741235..4675b9d0 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -4,6 +4,7 @@ import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.server.Server; import org.fusesource.hawtbuf.codec.StringCodec; diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 31f46ceb..916829e6 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -4,7 +4,7 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; -import java.io.Serializable; + import java.text.ParseException; import java.util.Collection; import java.util.HashMap; @@ -17,8 +17,9 @@ import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.proto.messages.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.ConnectionDescriptor; @@ -115,37 +116,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { disruptorPublish(new ProtocolEvent(session, msg)); } - /** - * Verify if the 2 topics matching respecting the rules of MQTT Appendix A - */ - protected boolean matchTopics(String msgTopic, String subscriptionTopic) { - try { - List msgTokens = SubscriptionsStore.splitTopic(msgTopic); - List subscriptionTokens = SubscriptionsStore.splitTopic(subscriptionTopic); - int i = 0; - for (; i< subscriptionTokens.size(); i++) { - Token subToken = subscriptionTokens.get(i); - if (subToken != Token.MULTI && subToken != Token.SINGLE) { - 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 - } - } - } - return i == msgTokens.size(); - } catch (ParseException ex) { - LOG.error(null, ex); - throw new RuntimeException(ex); - } - } - + /** * NOT SAFE Method, to be removed because used only in tests */ @@ -326,7 +297,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { for (final Subscription sub : subscriptions.matches(topic)) { if (qos == QOSType.MOST_ONE) { //QoS 0 - notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); + notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it @@ -335,7 +306,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), evt.getMessageID(), null); m_storageService.storePublishForFuture(newPublishEvt); } else { - notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); + notify(new NotifyEvent(sub.getClientId(), topic, qos, message, false)); } } } @@ -357,14 +328,14 @@ public class SimpleMessaging implements IMessaging, EventHandler { //scans retained messages to be published to the new subscription Collection messages = m_storageService.searchMatching(new IMatchingCondition() { public boolean match(String key) { - return matchTopics(key, topic); + return SubscriptionsStore.matchTopics(key, topic); } }); for (StoredMessage storedMsg : messages) { //fire the as retained the message LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); - notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(), storedMsg.getPayload(), true)); + notify(new NotifyEvent(newSubscription.getClientId(), topic, storedMsg.getQos(), storedMsg.getPayload(), true)); } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java index fba93dd8..189c5679 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java @@ -1,6 +1,6 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; -import org.dna.mqtt.moquette.messaging.spi.impl.Subscription; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; /** * diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java similarity index 97% rename from parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java rename to parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java index b065985c..6f2a1abb 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java @@ -1,4 +1,4 @@ -package org.dna.mqtt.moquette.messaging.spi.impl; +package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions; import java.io.Serializable; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java similarity index 90% rename from parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java rename to parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index d23872f1..2683cf10 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -1,18 +1,13 @@ -package org.dna.mqtt.moquette.messaging.spi.impl; +package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions; import java.text.ParseException; import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.Queue; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingQueue; import org.dna.mqtt.moquette.messaging.spi.IStorageService; -import org.fusesource.hawtbuf.codec.StringCodec; -import org.fusesource.hawtdb.api.BTreeIndexFactory; -import org.fusesource.hawtdb.api.MultiIndexFactory; -import org.fusesource.hawtdb.api.SortedIndex; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -350,6 +345,36 @@ public class SubscriptionsStore { return subscriptions.size(); } + /** + * Verify if the 2 topics matching respecting the rules of MQTT Appendix A + */ + public static boolean matchTopics(String msgTopic, String subscriptionTopic) { + try { + List msgTokens = SubscriptionsStore.splitTopic(msgTopic); + List subscriptionTokens = SubscriptionsStore.splitTopic(subscriptionTopic); + int i = 0; + for (; i< subscriptionTokens.size(); i++) { + Token subToken = subscriptionTokens.get(i); + if (subToken != Token.MULTI && subToken != Token.SINGLE) { + 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 + } + } + } + return i == msgTokens.size(); + } catch (ParseException ex) { + LOG.error(null, ex); + throw new RuntimeException(ex); + } + } protected static List splitTopic(String topic) throws ParseException { List res = new ArrayList(); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java index 5e1790d8..fd0de890 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import java.util.Collection; diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java index ee1ee312..5f6828cc 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java @@ -5,6 +5,7 @@ import org.apache.mina.core.session.DummySession; import org.apache.mina.core.session.IoSession; import org.apache.mina.core.write.WriteRequest; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import static org.junit.Assert.*; @@ -122,32 +123,6 @@ public class SimpleMessagingTest { //TODO check received message attributes } - @Test - public void testMatchTopics_simple() { - assertTrue(messaging.matchTopics("/", "/")); - assertTrue(messaging.matchTopics("/finance", "/finance")); - } - - @Test - public void testMatchTopics_multi() { - assertTrue(messaging.matchTopics("finance", "#")); - assertTrue(messaging.matchTopics("finance", "finance/#")); - assertTrue(messaging.matchTopics("finance/stock", "finance/#")); - assertTrue(messaging.matchTopics("finance/stock/ibm", "finance/#")); - } - - - @Test - public void testMatchTopics_single() { - assertTrue(messaging.matchTopics("finance", "+")); - assertTrue(messaging.matchTopics("finance/stock", "finance/+")); - assertTrue(messaging.matchTopics("/finance", "/+")); - assertFalse(messaging.matchTopics("/finance", "+")); - assertTrue(messaging.matchTopics("/finance", "+/+")); - assertTrue(messaging.matchTopics("/finance/stock/ibm", "/finance/+/ibm")); - assertFalse(messaging.matchTopics("/finance/stock", "+")); - } - @Test public void testHandleConnect_BadProtocol() { connMsg.setProcotolVersion((byte) 0x02); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java similarity index 85% rename from parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java rename to parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java index d9039890..c96397ae 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -1,12 +1,15 @@ -package org.dna.mqtt.moquette.messaging.spi.impl; +package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions; import java.io.File; import java.io.IOException; import java.text.ParseException; import java.util.Arrays; import java.util.List; -import org.dna.mqtt.moquette.MQTTException; -import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; + +import org.dna.mqtt.moquette.messaging.spi.impl.DummyStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore.Token; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.fusesource.hawtdb.api.MultiIndexFactory; import org.fusesource.hawtdb.api.PageFile; @@ -236,6 +239,32 @@ public class SubscriptionsStoreTest { assertEquals(1, store.size()); } + @Test + public void testMatchTopics_simple() { + assertTrue(SubscriptionsStore.matchTopics("/", "/")); + assertTrue(SubscriptionsStore.matchTopics("/finance", "/finance")); + } + + @Test + public void testMatchTopics_multi() { + assertTrue(SubscriptionsStore.matchTopics("finance", "#")); + assertTrue(SubscriptionsStore.matchTopics("finance", "finance/#")); + assertTrue(SubscriptionsStore.matchTopics("finance/stock", "finance/#")); + assertTrue(SubscriptionsStore.matchTopics("finance/stock/ibm", "finance/#")); + } + + + @Test + public void testMatchTopics_single() { + assertTrue(SubscriptionsStore.matchTopics("finance", "+")); + assertTrue(SubscriptionsStore.matchTopics("finance/stock", "finance/+")); + assertTrue(SubscriptionsStore.matchTopics("/finance", "/+")); + assertFalse(SubscriptionsStore.matchTopics("/finance", "+")); + assertTrue(SubscriptionsStore.matchTopics("/finance", "+/+")); + assertTrue(SubscriptionsStore.matchTopics("/finance/stock/ibm", "/finance/+/ibm")); + assertFalse(SubscriptionsStore.matchTopics("/finance/stock", "+")); + } + private static Token[] asArray(Object... l) { Token[] tokens = new Token[l.length];