diff --git a/ChangeLog.txt b/ChangeLog.txt index e30c92a3..aedec266 100644 --- a/ChangeLog.txt +++ b/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 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 3c90e13f..44b1ae41 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/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> m_persistentSubscriptions = new HashMap>(); - private Map m_retainedStore = new HashMap(); + private Map> m_persistentSubscriptions = new HashMap<>(); + private Map 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> m_persistentMessageStore = new HashMap>(); - private Map m_inflightStore = new HashMap(); + private Map> m_persistentMessageStore = new HashMap<>(); + private Map m_inflightStore = new HashMap<>(); private Map> m_inflightIDs = new HashMap<>(); - private Map m_qos2Store = new HashMap(); + private Map m_qos2Store = new HashMap<>(); private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 170b5fdf..f3a72616 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -221,25 +221,6 @@ class ProtocolProcessor implements EventHandler { 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 { /** * 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 { 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 { } } - 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); diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java index eefafd05..76a70a6e 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationPahoTest.java +++ b/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(); + } + } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 40fa58a3..248d1b57 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -695,4 +695,39 @@ public class ProtocolProcessorTest { }); assertTrue(messages.isEmpty()); } + + List 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()); + } }