diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java index 63676fe4..b9c4be98 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/INotifier.java @@ -8,6 +8,6 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; */ public interface INotifier { - public void notify(String clientId, String topic, QOSType qOSType, byte[] payload); + public void notify(String clientId, String topic, QOSType qOSType, byte[] payload, boolean retained); } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 7175f23f..e415f1b7 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -102,6 +102,15 @@ public class SimpleMessaging implements IMessaging { for (Subscription sub : subscriptions.matches(topic)) { m_notifier.notify(sub.clientId, topic, defQos, message, false); } + if (retain) { + if (message.length == 0) { + //clean the message from topic + m_retainedStore.remove(topic); + } else { + //store the message to the topic + m_retainedStore.put(topic, new StoredMessage(message, qos)); + } + } } finally { rwLock.readLock().unlock(); } @@ -111,6 +120,18 @@ public class SimpleMessaging implements IMessaging { Subscription newSubscription = new Subscription(clientId, topic, qos); rwLock.writeLock().lock(); subscriptions.add(newSubscription); + + //scans reatained messages to be published to the new subscription + LOG.debug("Scanning all retained messages..."); + for (Map.Entry entry : m_retainedStore) { + StoredMessage storedMsg = entry.getValue(); + if (matchTopics(entry.getKey(), topic)) { + //fire the as retained the message + m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), + storedMsg.getPayload(), true); + } + } + LOG.debug("Finished firing retained messages"); rwLock.writeLock().unlock(); }