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

Added cleanSession flag to Subscription class

Browse Source
This commit is contained in:
selva.andre
2012-09-01 10:57:59 +02:00
parent 3be2ccca8f
commit ba7f5284cf
7 changed files with 39 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
+1 -1
View File
@@ -28,7 +28,7 @@ public interface IMessaging {
/**
* Subscribe a client to a specified topic with a defined level
*/
void subscribe(String clientId, String topic, QOSType qos);
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession);
/**
* Remove all subscription to any topic the client (identified by clientID)
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+2 -2
View File
@@ -128,8 +128,8 @@ public class SimpleMessaging implements IMessaging, Runnable {
}
}
public void subscribe(String clientId, String topic, QOSType qos) {
Subscription newSubscription = new Subscription(clientId, topic, qos);
public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession) {
Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession);
try {
LOG.debug("subscribe invoked for topic: " + topic);
m_inboundQueue.put(new SubscribeEvent(newSubscription));
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/Subscription.java
+7 -1
View File
@@ -15,11 +15,13 @@ public class Subscription implements Serializable {
QOSType requestedQos;
String clientId;
String topic;
boolean cleanSession;
public Subscription(String clientId, String topic, QOSType requestedQos) {
public Subscription(String clientId, String topic, QOSType requestedQos, boolean cleanSession) {
this.requestedQos = requestedQos;
this.clientId = clientId;
this.topic = topic;
this.cleanSession = cleanSession;
}
public String getClientId() {
@@ -34,6 +36,10 @@ public class Subscription implements Serializable {
return topic;
}
public boolean isCleanSession() {
return this.cleanSession;
}
@Override
public boolean equals(Object obj) {
if (obj == null) {
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+4 -2
View File
@@ -147,8 +147,10 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
protected void handleSubscribe(IoSession session, SubscribeMessage msg) {
LOG.debug("handleSubscribe, registering the subscriptions");
for (SubscribeMessage.Couple req : msg.subscriptions()) {
m_messaging.subscribe((String) session.getAttribute(Constants.ATTR_CLIENTID),
req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()]);
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()],
cleanSession);
}
//ack the client
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+5 -5
View File
@@ -28,30 +28,30 @@ public class SimpleMessagingTest {
@Test
public void testSubscribe() {
//Exercise
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
messaging.processSubscribe(evt);
//Verify
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
assertTrue(messaging.getSubscriptions().contains(expectedSubscription));
}
@Test
public void testDoubleSubscribe() {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
messaging.processSubscribe(evt);
//Exercise
messaging.processSubscribe(evt);
//Verify
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE);
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
assertEquals(1, messaging.getSubscriptions().size());
}
@Test
public void testPublish() throws InterruptedException {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE));
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false));
messaging.processSubscribe(evt);
// INotifier notifier = mock(INotifier.class);
// messaging.setNotifier(notifier);
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java
+18 -18
View File
@@ -105,11 +105,11 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimple() {
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE);
Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashSub);
assertTrue(store.matches("finance").isEmpty());
Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE);
Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashFinanceSub);
assertTrue(store.matches("finance").isEmpty());
@@ -119,19 +119,19 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
assertTrue(store.matches("finance").contains(anySub));
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(financeAnySub);
assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub)));
}
@Test
public void testMatchingDeepMulti_one_layer() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
store.add(financeAnySub);
@@ -143,7 +143,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchingDeepMulti_two_layer() {
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE);
Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE, false);
store.add(financeAnySub);
//Verify
@@ -152,18 +152,18 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleSingle() {
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
assertTrue(store.matches("finance").contains(anySub));
Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE);
Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(financeOne);
assertTrue(store.matches("finance/stock").contains(financeOne));
}
@Test
public void testMatchManySingle() {
Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE);
Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(manySub);
//verify
@@ -173,9 +173,9 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSlashSingle() {
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashPlusSub);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE);
Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(anySub);
//Verify
@@ -187,10 +187,10 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchManyDeepSingle() {
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashPlusSub);
Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE);
Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE, false);
store.add(slashPlusDeepSub);
//Verify
@@ -199,7 +199,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti_allTheTree() {
store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false));
assertFalse(store.matches("finance").isEmpty());
assertFalse(store.matches("finance/ibm").isEmpty());
}
@@ -207,7 +207,7 @@ public class SubscriptionsStoreTest {
@Test
public void testMatchSimpleMulti_zeroLevel() {
//check MULTI in case of zero level match
store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false));
assertFalse(store.matches("finance").isEmpty());
}
@@ -215,7 +215,7 @@ public class SubscriptionsStoreTest {
@Test
public void testRemoveClientSubscriptions_existingClientID() {
String cliendID = "FAKE_CLID_1";
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false));
//Exercise
store.removeForClient(cliendID);
@@ -227,7 +227,7 @@ public class SubscriptionsStoreTest {
@Test
public void testRemoveClientSubscriptions_notexistingClientID() {
String cliendID = "FAKE_CLID_1";
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE));
store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false));
//Exercise
store.removeForClient("FAKE_CLID_2");
parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java
+2 -2
View File
@@ -100,7 +100,7 @@ public class MQTTHandlerTest {
IMessaging mockedMessaging = mock(IMessaging.class);
m_session.setAttribute(Constants.ATTR_CLIENTID, "fakeID");
m_session.setAttribute(Constants.CLEAN_SESSION, false);
//Exercise
m_handler.setMessaging(mockedMessaging);
@@ -108,7 +108,7 @@ public class MQTTHandlerTest {
//Verify
ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);
verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture());
verify(mockedMessaging).subscribe(eq("fakeID"), eq(topicName), argument.capture(), eq(false));
assertEquals(AbstractMessage.QOSType.EXACTLY_ONCE, argument.getValue());
//TODO verify the message ACK and corresponding QoS!!
}