Fixed bad qos publish, issue #56

ChangeLog.txt

@@ -21,6 +21,7 @@ Version 0.7:
     Fixed issue on multiple publish for overlapping subscription by the same client (fix issue #52).
     Moved the authorization check before dropping an already connected client with same clientID (fix issue #44).
     Fixed subscription tree that was unable to update clean session flag on subs with same client, topic and qos (fix issue #55)
+    Fixed overwriting of published qos in forward publish loop (fix issue #56)
 
 Version 0.6:
     Fix issue #42, bad password file path composition

broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java

@@ -34,13 +34,13 @@ import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
  */
 public class MemoryStorageService implements IMessagesStore, ISessionsStore {
     
-    private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<String, Set<Subscription>>();
+    private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<>();
-    private Map<String, StoredMessage> m_retainedStore = new HashMap<String, StoredMessage>();
+    private Map<String, StoredMessage> m_retainedStore = new HashMap<>();
     //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map)
-    private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<String, List<PublishEvent>>();
+    private Map<String, List<PublishEvent>> m_persistentMessageStore = new HashMap<>();
-    private Map<String, PublishEvent> m_inflightStore = new HashMap<String, PublishEvent>();
+    private Map<String, PublishEvent> m_inflightStore = new HashMap<>();
     private Map<String, Set<Integer>> m_inflightIDs = new HashMap<>();
-    private Map<String, PublishEvent> m_qos2Store = new HashMap<String, PublishEvent>();
+    private Map<String, PublishEvent> m_qos2Store = new HashMap<>();
     
     private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class);
     

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -221,25 +221,6 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
             m_willStore.put(msg.getClientID(), will);
         }
 
-//        //handle user authentication
-//        if (msg.isUserFlag()) {
-//            String pwd = null;
-//            if (msg.isPasswordFlag()) {
-//                pwd = msg.getPassword();
-//            } else if (!this.allowAnonymous) {
-//                failedCredentials(session);
-//                return;
-//            }
-//            if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
-//                failedCredentials(session);
-//                return;
-//            }
-//            session.setAttribute(NettyChannel.ATTR_KEY_USERNAME, msg.getUsername());
-//        } else if (!this.allowAnonymous) {
-//            failedCredentials(session);
-//            return;
-//        }
-
         subscriptions.activate(msg.getClientID());
 
         //handle clean session flag
@@ -381,9 +362,9 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     /**
      * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
      * */
-    private void forward2Subscribers(PublishEvent pubEvt) {
+    void forward2Subscribers(PublishEvent pubEvt) {
         final String topic = pubEvt.getTopic();
-        AbstractMessage.QOSType qos = pubEvt.getQos();
+        final AbstractMessage.QOSType publishingQos = pubEvt.getQos();
         final ByteBuffer origMessage = pubEvt.getMessage();
         boolean retain = pubEvt.isRetain();
         final Integer messageID = pubEvt.getMessageID();
@@ -393,6 +374,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
             LOG.debug("subscription tree {}", subscriptions.dumpTree());
         }
         for (final Subscription sub : subscriptions.matches(topic)) {
+            AbstractMessage.QOSType qos = publishingQos;
             if (qos.ordinal() > sub.getRequestedQos().ordinal()) {
                 qos = sub.getRequestedQos();
             }
@@ -428,7 +410,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         }
     }
 
-    private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) {
+    protected void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) {
         LOG.debug("sendPublish invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID);
         PublishMessage pubMessage = new PublishMessage();
         pubMessage.setRetainFlag(retained);

broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java

@@ -17,6 +17,7 @@ package org.eclipse.moquette.server;
 
 import org.eclipse.paho.client.mqttv3.*;
 import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
+import org.fusesource.mqtt.client.QoS;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -361,4 +362,47 @@ public class ServerIntegrationPahoTest {
         assertEquals("Hello MQTT 2", message.toString());
     }
 
+    /**
+     * subscriber A connect and subscribe on "a/b" QoS 1
+     * subscriber B connect and subscribe on "a/+"  BUT with QoS 2
+     * publisher connects and send a message "hello" on "a/b"
+     * subscriber A must receive a notification with QoS1
+     * subscriber B must receive a notification with QoS2
+     */
+    @Test
+    public void checkSubscribersGetCorrectQosNotifications() throws Exception {
+        LOG.info("*** checkSubscribersGetCorrectQosNotifications ***");
+        String tmpDir = System.getProperty("java.io.tmpdir");
+
+        MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberA");
+
+        MqttClient subscriberA = new MqttClient("tcp://localhost:1883", "SubscriberA", dsSubscriberA);
+        TestCallback cbSubscriberA = new TestCallback();
+        subscriberA.setCallback(cbSubscriberA);
+        subscriberA.connect();
+        subscriberA.subscribe("a/b", 1);
+
+        MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberB");
+
+        MqttClient subscriberB = new MqttClient("tcp://localhost:1883", "SubscriberB", dsSubscriberB);
+        TestCallback cbSubscriberB = new TestCallback();
+        subscriberB.setCallback(cbSubscriberB);
+        subscriberB.connect();
+        subscriberB.subscribe("a/+", 2);
+
+
+        m_client.connect();
+        m_client.publish("a/b", "Hello world MQTT!!".getBytes(), 2, false);
+
+        MqttMessage messageOnA = cbSubscriberA.getMessage(true);
+        assertEquals("Hello world MQTT!!", new String(messageOnA.getPayload()));
+        assertEquals(1, messageOnA.getQos());
+        subscriberA.disconnect();
+
+        MqttMessage messageOnB = cbSubscriberB.getMessage(true);
+        assertEquals("Hello world MQTT!!", new String(messageOnB.getPayload()));
+        assertEquals(2, messageOnB.getQos());
+        subscriberB.disconnect();
+    }
+
 }

broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java

@@ -695,4 +695,39 @@ public class ProtocolProcessorTest {
         });
         assertTrue(messages.isEmpty());
     }
+
+    List<PublishEvent> publishedForwarded = new ArrayList<>();
+
+    @Test
+    public void testForwardPublishWithCorrectQos() {
+        ByteBuffer payload = ByteBuffer.wrap("Hello world MQTT!!".getBytes());
+        PublishEvent forwardPublish = new PublishEvent("a/b", QOSType.EXACTLY_ONCE, payload, true, "Publisher", 1);
+        IMessagesStore memoryMessageStore = new MemoryStorageService();
+
+        Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false);
+        Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false);
+        SubscriptionsStore subscriptions = new SubscriptionsStore();
+        subscriptions.add(subQos1);
+        subscriptions.add(subQos2);
+
+
+        ProtocolProcessor processor = new ProtocolProcessor() {
+            @Override
+            protected void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message,
+                                       boolean retained, Integer messageID) {
+                publishedForwarded.add(new PublishEvent(topic, qos, message, retained, clientId, messageID));
+            }
+        };
+        processor.init(subscriptions, memoryMessageStore, null, null, true, null);
+
+        //Exercise
+        processor.forward2Subscribers(forwardPublish);
+
+        //Verify
+        assertEquals(2, publishedForwarded.size());
+        assertEquals(subQos1.getClientId(), publishedForwarded.get(0).getClientID());
+        assertEquals(subQos1.getRequestedQos(), publishedForwarded.get(0).getQos());
+        assertEquals(subQos2.getClientId(), publishedForwarded.get(1).getClientID());
+        assertEquals(subQos2.getRequestedQos(), publishedForwarded.get(1).getQos());
+    }
 }