From 3355e7c334e775394f26c8892957cd1de317b854 Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Fri, 9 Dec 2016 13:23:09 +0100
Subject: [PATCH] Moved selection of subscriptions out of route2Subscribers

---
 .../moquette/spi/impl/ProtocolProcessor.java  | 22 ++++++++++++-------
 .../spi/impl/ProtocolProcessorTest.java       |  8 ++++---
 2 files changed, 19 insertions(+), 11 deletions(-)

diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
index e0841db7..ec03476a 100644
--- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
@@ -461,7 +461,9 @@ public class ProtocolProcessor {
         IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
         String clientID = NettyUtils.clientID(channel);
         toStoreMsg.setClientID(clientID);
-        route2Subscribers(toStoreMsg);
+
+        List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
+        route2Subscribers(toStoreMsg, topicMatchingSubscriptions);
 
         if (msg.isRetainFlag()) {
             //QoS == 0 && retain => clean old retained
@@ -494,7 +496,8 @@ public class ProtocolProcessor {
         IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
         String clientID = NettyUtils.clientID(channel);
         toStoreMsg.setClientID(clientID);
-        route2Subscribers(toStoreMsg);
+        List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
+        route2Subscribers(toStoreMsg, topicMatchingSubscriptions);
 
         //send PUBACK
         final Integer messageID = msg.getMessageID();
@@ -573,7 +576,8 @@ public class ProtocolProcessor {
         if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
             guid = m_messagesStore.storePublishForFuture(toStoreMsg);
         }
-        route2Subscribers(toStoreMsg);
+        List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
+        route2Subscribers(toStoreMsg, topicMatchingSubscriptions);
 
         if (!msg.isRetainFlag()) {
             return;
@@ -604,14 +608,16 @@ public class ProtocolProcessor {
         IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
         tobeStored.setClientID(clientID);
         tobeStored.setMessageID(messageId);
-        route2Subscribers(tobeStored);
+        String topic = tobeStored.getTopic();
+        List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
+        route2Subscribers(tobeStored, topicMatchingSubscriptions);
     }
 
 
     /**
      * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
      * */
-    void route2Subscribers(IMessagesStore.StoredMessage pubMsg) {
+    void route2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
         final String topic = pubMsg.getTopic();
         final AbstractMessage.QOSType publishingQos = pubMsg.getQos();
         final ByteBuffer origMessage = pubMsg.getMessage();
@@ -626,7 +632,6 @@ public class ProtocolProcessor {
             guid = m_messagesStore.storePublishForFuture(pubMsg);
         }
 
-        List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
         LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic);
         for (final Subscription sub : topicMatchingSubscriptions) {
             AbstractMessage.QOSType qos = publishingQos;
@@ -745,10 +750,11 @@ public class ProtocolProcessor {
         LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
         ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
         IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
-        route2Subscribers(evt);
+        final String topic = evt.getTopic();
+        List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
+        route2Subscribers(evt, topicMatchingSubscriptions);
 
         if (evt.isRetained()) {
-            final String topic = evt.getTopic();
             if (!evt.getMessage().hasRemaining()) {
                 m_messagesStore.cleanRetained(topic);
             } else {
diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
index 6e674ab8..b6910740 100644
--- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
@@ -452,7 +452,8 @@ public class ProtocolProcessorTest {
 
     @Test
     public void testForwardPublishWithCorrectQos() {
-        StoredMessage forwardPublish = new StoredMessage("Hello world MQTT!!".getBytes(), QOSType.EXACTLY_ONCE, "a/b");
+        String topic = "a/b";
+        StoredMessage forwardPublish = new StoredMessage("Hello world MQTT!!".getBytes(), QOSType.EXACTLY_ONCE, topic);
         forwardPublish.setRetained(true);
         forwardPublish.setMessageID(1);
 
@@ -463,7 +464,7 @@ public class ProtocolProcessorTest {
         sessionsStore.createNewSession("Sub A", false);
         sessionsStore.createNewSession("Sub B", false);
 
-        Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE);
+        Subscription subQos1 = new Subscription("Sub A", topic, QOSType.LEAST_ONE);
         Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE);
         sessionsStore.addNewSubscription(subQos1);
         sessionsStore.addNewSubscription(subQos2);
@@ -490,7 +491,8 @@ public class ProtocolProcessorTest {
         processor.connectionDescriptors.put("Sub B", new ConnectionDescriptor("Sub B", null, true));
 
         //Exercise
-        processor.route2Subscribers(forwardPublish);
+        List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
+        processor.route2Subscribers(forwardPublish, topicMatchingSubscriptions);
 
         //Verify
         assertEquals(2, publishedForwarded.size());