From a62f0aa6f372bda4f4c0b7195d3652f7acb72891 Mon Sep 17 00:00:00 2001 From: Dolychuck Date: Sat, 5 Aug 2017 12:02:49 -0700 Subject: [PATCH] Fix issue #171 --- .../moquette/spi/impl/ProtocolProcessor.java | 23 +++++++++++-------- 1 file changed, 14 insertions(+), 9 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 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()) {