From b7f3e1beba3ff0ec2c933af5c110f6ef2a651446 Mon Sep 17 00:00:00 2001 From: Andreas Mager Date: Wed, 1 Feb 2017 15:55:47 +0100 Subject: [PATCH] broker/Qos1PublishHandler: code formatter used --- .../moquette/spi/impl/Qos1PublishHandler.java | 61 +++++++++++-------- 1 file changed, 37 insertions(+), 24 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index d9b72a58..8b0bf075 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -1,3 +1,4 @@ + package io.moquette.spi.impl; import io.moquette.server.ConnectionDescriptorStore; @@ -14,14 +15,13 @@ import io.netty.handler.codec.mqtt.MqttPubAckMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.util.List; - import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; class Qos1PublishHandler extends QosPublishHandler { + private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); private final SubscriptionsStore subscriptions; @@ -31,10 +31,9 @@ class Qos1PublishHandler extends QosPublishHandler { private final MessagesPublisher publisher; public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, - IMessagesStore messagesStore, BrokerInterceptor interceptor, - ConnectionDescriptorStore connectionDescriptors, - String brokerPort, MessagesPublisher messagesPublisher) { - super(authorizator); + IMessagesStore messagesStore, BrokerInterceptor interceptor, + ConnectionDescriptorStore connectionDescriptors, String brokerPort, MessagesPublisher messagesPublisher) { + super(authorizator); this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; @@ -43,7 +42,7 @@ class Qos1PublishHandler extends QosPublishHandler { } void receivedPublishQos1(Channel channel, MqttPublishMessage msg) { - //verify if topic can be write + // verify if topic can be write final String topic = msg.variableHeader().topicName(); if (checkWriteOnTopic(topic, channel)) { return; @@ -51,33 +50,41 @@ class Qos1PublishHandler extends QosPublishHandler { final int messageID = msg.variableHeader().messageId(); - //route message to subscribers + // route message to subscribers IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); 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); + } List topicMatchingSubscriptions = subscriptions.matches(topic); this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); - //send PUBACK - //TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { - sendPubAck(clientID, messageID); -// } + // send PUBACK + // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { + sendPubAck(clientID, messageID); + // } if (msg.fixedHeader().isRetain()) { if (!msg.payload().isReadable()) { m_messagesStore.cleanRetained(topic); } else { - //before wasn't stored + // before wasn't stored MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); m_messagesStore.storeRetained(topic, guid); } @@ -94,14 +101,20 @@ class Qos1PublishHandler extends QosPublishHandler { try { if (connectionDescriptors == null) { - throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!"); + throw new RuntimeException( + "Internal bad error, found connectionDescriptors to null while it should be initialized, " + + "somewhere it's overwritten!!"); } LOG.debug("clientIDs are {}", connectionDescriptors); if (!connectionDescriptors.isConnected(clientId)) { - throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors)); + throw new RuntimeException( + String.format( + "Can't find a ConnectionDescriptor for client %s in cache %s", + clientId, + connectionDescriptors)); } connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId); - } catch(Throwable t) { + } catch (Throwable t) { LOG.error(null, t); } }