diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index c020aa73..de76ada6 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -1,3 +1,4 @@ + package io.moquette.spi.impl; import io.moquette.server.ConnectionDescriptorStore; @@ -13,15 +14,13 @@ import io.netty.channel.Channel; import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.util.List; - import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; import static io.moquette.spi.impl.Utils.messageId; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; -class Qos2PublishHandler extends QosPublishHandler { +class Qos2PublishHandler extends QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); @@ -33,10 +32,10 @@ class Qos2PublishHandler extends QosPublishHandler { private final MessagesPublisher publisher; public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, - IMessagesStore messagesStore, BrokerInterceptor interceptor, - ConnectionDescriptorStore connectionDescriptors, - ISessionsStore sessionsStore, String brokerPort, MessagesPublisher messagesPublisher) { - super(authorizator); + IMessagesStore messagesStore, BrokerInterceptor interceptor, + ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, String brokerPort, + MessagesPublisher messagesPublisher) { + super(authorizator); this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; @@ -47,7 +46,7 @@ class Qos2PublishHandler extends QosPublishHandler { void receivedPublishQos2(Channel channel, MqttPublishMessage msg) { final String topic = msg.variableHeader().topicName(); - //check if the topic can be wrote + // check if the topic can be wrote if (checkWriteOnTopic(topic, channel)) { return; } @@ -57,23 +56,30 @@ class Qos2PublishHandler extends QosPublishHandler { String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - if (LOG.isTraceEnabled()) { - LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, " + - "payload = {}, subscriptionTree = {}.", - clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), - subscriptions.dumpTree()); - } else { - LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", - clientID, topic, messageID); - } + if (LOG.isTraceEnabled()) { + LOG.trace( + "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, " + + "payload = {}, subscriptionTree = {}.", + clientID, + topic, + messageID, + DebugUtils.payload2Str(toStoreMsg.getMessage()), + subscriptions.dumpTree()); + } else { + LOG.info( + "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", + clientID, + topic, + messageID); + } - //QoS2 + // QoS2 MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); - //TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { - sendPubRec(clientID, messageID); - //} - //Next the client will send us a pub rel - //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher + // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { + sendPubRec(clientID, messageID); + // } + // Next the client will send us a pub rel + // NB publish to subscribers for QoS 2 happen upon PUBREL from publisher if (msg.fixedHeader().isRetain()) { if (msg.payload().readableBytes() == 0) { @@ -87,9 +93,9 @@ class Qos2PublishHandler extends QosPublishHandler { } /** - * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish - * to all interested subscribers. - * */ + * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored + * message and publish to all interested subscribers. + */ void processPubRel(Channel channel, MqttMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); @@ -99,14 +105,22 @@ class Qos2PublishHandler extends QosPublishHandler { final String topic = evt.getTopic(); List topicMatchingSubscriptions = subscriptions.matches(topic); - if (LOG.isDebugEnabled()) { - LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", - clientID, topic, messageID, DebugUtils.payload2Str(evt.getMessage()), - subscriptions.dumpTree()); - } else { - LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID, - topic, messageID); - } + if (LOG.isDebugEnabled()) { + LOG.debug( + "Sending publish message to subscribers. " + + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + clientID, + topic, + messageID, + DebugUtils.payload2Str(evt.getMessage()), + subscriptions.dumpTree()); + } else { + LOG.info( + "Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", + clientID, + topic, + messageID); + } this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions); if (evt.isRetained()) { @@ -120,12 +134,11 @@ class Qos2PublishHandler extends QosPublishHandler { sendPubComp(clientID, messageID); } - private void sendPubRec(String clientID, int messageID) { LOG.debug("Sending PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID); MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, AT_MOST_ONCE, false, 0); MqttMessage pubRecMessage = new MqttMessage(fixedHeader, from(messageID)); - connectionDescriptors.sendMessage(pubRecMessage,messageID, clientID); + connectionDescriptors.sendMessage(pubRecMessage, messageID, clientID); } private void sendPubComp(String clientID, int messageID) {