From 7dece38b57c2217db64bd061335d7f604a4384c1 Mon Sep 17 00:00:00 2001 From: andrea Date: Tue, 23 Jun 2015 23:23:38 +0200 Subject: [PATCH] Fixed multiple publish notification for overlapping subscriptions (fix issue #52) --- ChangeLog.txt | 1 + .../subscriptions/SubscriptionsStore.java | 17 +++++- .../server/ServerIntegrationFuseTest.java | 37 ++++++++++++ .../subscriptions/SubscriptionsStoreTest.java | 19 ++++++- .../issue58_overlappingSubscription.groovy | 56 +++++++++++++++++++ 5 files changed, 124 insertions(+), 6 deletions(-) create mode 100644 tools_scripts/integration/issue58_overlappingSubscription.groovy diff --git a/ChangeLog.txt b/ChangeLog.txt index 8819482b..aad71d4d 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -18,6 +18,7 @@ Version 0.7: Bad assignment of PacketID during publish to subscribed QoS0 topics (fix issue #38) Fixed duplicated subscription with different QoS same client and topic (fix issue #49) Fixed UNSUBSCRIBE that missed to remove the subscription from the persistent store (fix issue #50) + Fixed issue on multiple publish for overlapping subscription by the same client (fix issue #52). Version 0.6: Fix issue #42, bad password file path composition diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 2404b8c4..e76da16c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -67,6 +67,7 @@ public class SubscriptionsStore { String s = ""; for (int i=0; i < deep; i++) { s += "\t"; +// s += "--"; } return s; } @@ -225,10 +226,20 @@ public class SubscriptionsStore { return Collections.emptyList(); } - Queue tokenQueue = new LinkedBlockingDeque(tokens); - List matchingSubs = new ArrayList(); + Queue tokenQueue = new LinkedBlockingDeque<>(tokens); + List matchingSubs = new ArrayList<>(); subscriptions.matches(tokenQueue, matchingSubs); - return matchingSubs; + + //remove the overlapping subscriptions, selecting ones with greatest qos + Map subsForClient = new HashMap<>(); + for (Subscription sub : matchingSubs) { + Subscription existingSub = subsForClient.get(sub.getClientId()); + //update the selected subscriptions if not present or if has a greater qos + if (existingSub == null || existingSub.getRequestedQos().ordinal() < sub.getRequestedQos().ordinal()) { + subsForClient.put(sub.getClientId(), sub); + } + } + return /*matchingSubs*/new ArrayList<>(subsForClient.values()); } public boolean contains(Subscription sub) { diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java index dc48254a..db6b791f 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java @@ -245,4 +245,41 @@ public class ServerIntegrationFuseTest { msg.ack(); assertEquals("Hello2", new String(msg.getPayload())); } + + /** + * subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 + * publisher connects and send a message "hello" on "a/b" + * subscriber must receive only a single message not twice + */ + @Test + public void checkSinglePublishOnOverlappingSubscriptions() throws Exception { + LOG.info("*** checkSinglePublishOnOverlappingSubscriptions ***"); + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", 1883); + mqtt.setClientId("Publisher"); + m_publisher = mqtt.blockingConnection(); + m_publisher.connect(); + + m_mqtt.setHost("localhost", 1883); + m_mqtt.setCleanSession(false); + m_mqtt.setClientId("Subscriber"); + m_subscriber = m_mqtt.blockingConnection(); + m_subscriber.connect(); + Topic[] topics = new Topic[]{ + new Topic("a/+", QoS.EXACTLY_ONCE), + new Topic("a/b", QoS.AT_LEAST_ONCE)}; + m_subscriber.subscribe(topics); + + //force the publisher to send + m_publisher.publish("a/b", "Hello world MQTT!!".getBytes(), QoS.AT_LEAST_ONCE, false); + + //reconnect and expect to receive the hello 2 message + Message msg = m_subscriber.receive(1, TimeUnit.SECONDS); + assertNotNull(msg); + msg.ack(); + assertEquals("Hello world MQTT!!", new String(msg.getPayload())); + //try to listen a second publish + msg = m_subscriber.receive(1, TimeUnit.SECONDS); + assertNull(msg); + } } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 72a8a8b1..ef8ebaa1 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -117,7 +117,7 @@ public class SubscriptionsStoreTest { store.add(anySub); assertTrue(store.matches("finance").contains(anySub)); - Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); store.add(financeAnySub); assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub))); } @@ -125,7 +125,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchingDeepMulti_one_layer() { 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); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); store.add(anySub); store.add(financeAnySub); @@ -184,7 +184,7 @@ public class SubscriptionsStoreTest { 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, false); + Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_2", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE, false); store.add(slashPlusDeepSub); //Verify @@ -270,6 +270,19 @@ public class SubscriptionsStoreTest { assertEquals(1, store.size()); } + @Test + public void testOverlappingSubscriptions() { + Subscription genericSub = new Subscription("FAKE_CLI_ID_1", "a/+", AbstractMessage.QOSType.EXACTLY_ONCE, false); + store.add(genericSub); + Subscription specificSub = new Subscription("FAKE_CLI_ID_1", "a/b", AbstractMessage.QOSType.LEAST_ONE, false); + store.add(specificSub); + + //Verify + assertEquals(1, store.matches("a/b").size()); + //assertTrue(store.matches("/finance").contains(slashPlusSub)); + //assertFalse(store.matches("/finance").contains(anySub)); + } + @Test public void testMatchTopics_simple() { assertTrue(SubscriptionsStore.matchTopics("/", "/")); diff --git a/tools_scripts/integration/issue58_overlappingSubscription.groovy b/tools_scripts/integration/issue58_overlappingSubscription.groovy new file mode 100644 index 00000000..9aff5f6e --- /dev/null +++ b/tools_scripts/integration/issue58_overlappingSubscription.groovy @@ -0,0 +1,56 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.10') + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +import java.util.concurrent.TimeUnit + +String host = args.size() == 1 ? args[0] : "localhost" + +MQTT mqttSub = new MQTT() +mqttSub.setHost(host, 1883) +mqttSub.setClientId("Subscriber") +mqttSub.cleanSession = false +//AT_MOST_ONCE, +//AT_LEAST_ONCE, +//EXACTLY_ONCE +BlockingConnection clientSub = mqttSub.blockingConnection() +clientSub.connect() +println "Sub connected" +Topic[] overlappingSubscriptions = [new Topic("a/+", QoS.EXACTLY_ONCE), + new Topic("a/b", QoS.AT_LEAST_ONCE)] +byte[] qoses = clientSub.subscribe(overlappingSubscriptions) +println "Sub subscribed" + +MQTT mqttPub = new MQTT() +mqttPub.setHost(host, 1883) +mqttPub.setClientId("Publisher") +mqttPub.cleanSession = false + +BlockingConnection clientPub = mqttPub.blockingConnection() +clientPub.connect() +println "Pub connected" +clientPub.publish("a/b", "Hello world".bytes, QoS.AT_LEAST_ONCE, false) +println "Pub published" +Message msg = clientSub.receive() +println "Sub received message \n\t Topic: ${msg.topic}, payload: ${new String(msg.payload)}" +msg.ack() + +//try to receive another NOT want publish +msg = clientSub.receive(1, TimeUnit.SECONDS) +if (msg != null) { + println "Second possible message, payload: ${new String(msg.payload)}, topic: ${msg.topic}" +} + +clientPub.disconnect() +clientPub.kill() +println "Pub disconnected" + +clientSub.disconnect() +clientSub.kill() +println "Sub disconnected" + +println "Finished!!\n\n"