diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index 9689d801..0a58c8fa 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java @@ -215,7 +215,12 @@ public final class Client { * It's admitted to publish a 0 -length payload. */ public void publish(String topic, byte[] payload) throws PublishException { + publish(topic, payload, false); + } + + public void publish(String topic, byte[] payload, boolean retain) throws PublishException { PublishMessage msg = new PublishMessage(); + msg.setRetainFlag(retain); msg.setTopicName(topic); msg.setPayload(payload); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java b/parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java index 0fdede1b..c8593b52 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java @@ -6,5 +6,5 @@ package org.dna.mqtt.moquette.client; */ public interface IPublishCallback { - void published(String topic, byte[] message); + void published(String topic, byte[] message/*, boolean retained*/); } 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 c71b90e4..0c196613 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 @@ -199,8 +199,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { m_authenticator = authenticator; } - public void notify(String clientId, String topic, QOSType qOSType, byte[] payload) { + public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained) { PublishMessage pubMessage = new PublishMessage(); + pubMessage.setRetainFlag(retained); pubMessage.setTopicName(topic); pubMessage.setQos(qOSType); pubMessage.setPayload(payload); 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 93a5ab98..b90a8862 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 @@ -56,7 +56,33 @@ public class SimpleMessagingTest { //Verify ArgumentCaptor argument = ArgumentCaptor.forClass(byte[].class); - verify(notifier).notify(eq(FAKE_CLIENT_ID), eq(FAKE_TOPIC), any(QOSType.class), argument.capture()); + verify(notifier).notify(eq(FAKE_CLIENT_ID), eq(FAKE_TOPIC), any(QOSType.class), argument.capture(), eq(false)); assertEquals("Hello", new String(argument.getValue())); } + + @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", "+")); + } } 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 039156ce..e978cee9 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 @@ -1,10 +1,17 @@ package org.dna.mqtt.moquette.messaging.spi.impl; +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.proto.messages.AbstractMessage; +import org.fusesource.hawtdb.api.MultiIndexFactory; +import org.fusesource.hawtdb.api.PageFile; +import org.fusesource.hawtdb.api.PageFileFactory; +import org.junit.After; import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; @@ -14,15 +21,33 @@ import org.junit.Test; * @author andrea */ public class SubscriptionsStoreTest { - + + private File hawtFile; private SubscriptionsStore store; public SubscriptionsStoreTest() { } @Before - public void setUp() { + public void setUp() throws IOException { + hawtFile = File.createTempFile("moquette_persistent_store", ".dat"); + store = new SubscriptionsStore(); +// store.init(hawtFile.getAbsolutePath()); + PageFileFactory pageFactory = new PageFileFactory(); + + pageFactory.setFile(hawtFile); + pageFactory.open(); + PageFile pageFile = pageFactory.getPageFile(); + MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile); + store.init(multiIndexFactory); + } + + @After + public void tearDown() { + if (hawtFile.exists()) { + hawtFile.delete(); + } } @Test @@ -80,11 +105,11 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimple() { - Subscription slashSub = new Subscription(null, "/", AbstractMessage.QOSType.MOST_ONE); + Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE); store.add(slashSub); assertTrue(store.matches("finance").isEmpty()); - Subscription slashFinanceSub = new Subscription(null, "/finance", AbstractMessage.QOSType.MOST_ONE); + Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE); store.add(slashFinanceSub); assertTrue(store.matches("finance").isEmpty()); @@ -94,19 +119,19 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti() { - Subscription anySub = new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE); store.add(anySub); assertTrue(store.matches("finance").contains(anySub)); - Subscription financeAnySub = new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE); store.add(financeAnySub); assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub))); } @Test public void testMatchingDeepMulti_one_layer() { - Subscription anySub = new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE); - Subscription financeAnySub = new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE); + 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); store.add(anySub); store.add(financeAnySub); @@ -118,7 +143,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchingDeepMulti_two_layer() { - Subscription financeAnySub = new Subscription(null, "finance/stock/#", AbstractMessage.QOSType.MOST_ONE); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE); store.add(financeAnySub); //Verify @@ -127,18 +152,18 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleSingle() { - Subscription anySub = new Subscription(null, "+", AbstractMessage.QOSType.MOST_ONE); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE); store.add(anySub); assertTrue(store.matches("finance").contains(anySub)); - Subscription financeOne = new Subscription(null, "finance/+", AbstractMessage.QOSType.MOST_ONE); + Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE); store.add(financeOne); assertTrue(store.matches("finance/stock").contains(financeOne)); } @Test public void testMatchManySingle() { - Subscription manySub = new Subscription(null, "+/+", AbstractMessage.QOSType.MOST_ONE); + Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE); store.add(manySub); //verify @@ -148,9 +173,9 @@ public class SubscriptionsStoreTest { @Test public void testMatchSlashSingle() { - Subscription slashPlusSub = new Subscription(null, "/+", AbstractMessage.QOSType.MOST_ONE); + Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE); store.add(slashPlusSub); - Subscription anySub = new Subscription(null, "+", AbstractMessage.QOSType.MOST_ONE); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE); store.add(anySub); //Verify @@ -162,10 +187,10 @@ public class SubscriptionsStoreTest { @Test public void testMatchManyDeepSingle() { - Subscription slashPlusSub = new Subscription(null, "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE); + Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE); store.add(slashPlusSub); - Subscription slashPlusDeepSub = new Subscription(null, "/+/stock/+", AbstractMessage.QOSType.MOST_ONE); + Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_1", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE); store.add(slashPlusDeepSub); //Verify @@ -174,7 +199,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_allTheTree() { - store.add(new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE)); + store.add(new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE)); assertFalse(store.matches("finance").isEmpty()); assertFalse(store.matches("finance/ibm").isEmpty()); } @@ -182,7 +207,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_zeroLevel() { //check MULTI in case of zero level match - store.add(new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE)); + store.add(new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE)); assertFalse(store.matches("finance").isEmpty()); } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 8f38e4eb..50592f73 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java @@ -196,4 +196,33 @@ public class ServerIntegrationTest { client.shutdown(); } + + +// @Test +// public void testRetain_maintainMessage_againstClientDestruction() { +// Client client = new Client("localhost", Server.PORT, "CLID_123"); +// client.connect(false); //without session cleanup +// +// client.publish("/topic", "Test my payload".getBytes(), true); +// client.close(); +// client.shutdown(); +// +// //Exercise that the client maintain the subscriptions +// client = new Client("localhost", Server.PORT, "CLID_123"); +// client.connect(false); +// client.subscribe("/topic", new IPublishCallback() { +// +// public void published(String topic, byte[] message) { +// received = true; +// } +// }); +// client.close(); +// +// //Verify +// assertTrue(received); +// +// //TearDown +// client.shutdown(); +// } + }