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

Moved SubscriptionStore and its stuff in separate package

Browse Source
This commit is contained in:
selva.andre
2012-09-05 21:26:53 +02:00
parent 2f67532fdd
commit 562c99b9ed
9 changed files with 77 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java
+1 -2
View File
@@ -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;
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+1
View File
@@ -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;
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+8 -37
View File
@@ -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<ValueEvent> {
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<Token> msgTokens = SubscriptionsStore.splitTopic(msgTopic);
List<Token> 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<ValueEvent> {
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<ValueEvent> {
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<ValueEvent> {
//scans retained messages to be published to the new subscription
Collection<StoredMessage> 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));
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/SubscribeEvent.java
+1 -1
View File
@@ -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;
/**
*
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java → parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/Subscription.java
+1 -1
View File
@@ -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;
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java → parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java
+31 -6
View File
@@ -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<Token> msgTokens = SubscriptionsStore.splitTopic(msgTopic);
List<Token> 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<Token> splitTopic(String topic) throws ParseException {
List res = new ArrayList<Token>();
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java
+1
View File
@@ -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;
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+1 -26
View File
@@ -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);
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java → parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java
+32 -3
View File
@@ -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];