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 cd099cf6..49d8de0a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -561,15 +561,20 @@ public class ProtocolProcessor { /** * Specialized version to publish will testament message. */ - private void forwardPublishWill(WillMessage will, String clientID) { - LOG.info("Publishing will message. CId={}, topic={}", clientID, will.getTopic()); - // it has just to publish the message downstream to the subscribers - // NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1 - IMessagesStore.StoredMessage tobeStored = asStoredMessage(will); - tobeStored.setClientID(clientID); - Topic topic = new Topic(tobeStored.getTopic()); - this.messagesPublisher.publish2Subscribers(tobeStored, topic); - } + private void forwardPublishWill(WillMessage will, String clientID) { + LOG.info("Publishing will message. CId={}, topic={}", clientID, will.getTopic()); + // it has just to publish the message downstream to the subscribers + // NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1 + IMessagesStore.StoredMessage tobeStored = asStoredMessage(will); + tobeStored.setClientID(clientID); + Topic topic = new Topic(tobeStored.getTopic()); + this.messagesPublisher.publish2Subscribers(tobeStored, topic); + + //Stores retained message to the topic + if(will.isRetained()) { + m_messagesStore.storeRetained(topic, tobeStored); + } + } static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) { if (qos.value() > sub.getRequestedQos().value()) {