From ba7f5284cf79200ec1056b1d65bbe8944aa6fab6 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 10:57:59 +0200 Subject: [PATCH] Added cleanSession flag to Subscription class --- .../moquette/messaging/spi/IMessaging.java | 2 +- .../messaging/spi/impl/SimpleMessaging.java | 4 +-- .../messaging/spi/impl/Subscription.java | 8 ++++- .../dna/mqtt/moquette/server/MQTTHandler.java | 6 ++-- .../spi/impl/SimpleMessagingTest.java | 10 +++--- .../spi/impl/SubscriptionsStoreTest.java | 36 +++++++++---------- .../mqtt/moquette/server/MQTTHandlerTest.java | 4 +-- 7 files changed, 39 insertions(+), 31 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 733f0615..4f86fb44 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -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) 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 7e97b652..b0a4dde3 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 @@ -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)); 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/Subscription.java index 83fab74e..59f431ac 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/Subscription.java @@ -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) { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 727f8672..cd2f5126 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -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 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 38083ba6..404aebf7 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 @@ -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); 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/SubscriptionsStoreTest.java index e978cee9..01a02960 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/SubscriptionsStoreTest.java @@ -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"); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java index 2c143703..dd227bf3 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/MQTTHandlerTest.java @@ -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 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!! }