Fixed multiple publish notification for overlapping subscriptions (fix issue #52)

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

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<Token> tokenQueue = new LinkedBlockingDeque<Token>(tokens);
+        Queue<Token> tokenQueue = new LinkedBlockingDeque<>(tokens);
-        List<Subscription> matchingSubs = new ArrayList<Subscription>();
+        List<Subscription> matchingSubs = new ArrayList<>();
         subscriptions.matches(tokenQueue, matchingSubs);
-        return matchingSubs;
+
+        //remove the overlapping subscriptions, selecting ones with greatest qos
+        Map<String, Subscription> 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) {

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);
+    }
 }

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("/", "/"));

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"