From 91ca7fd1c22faf3babae1afc1869430f14fe7e43 Mon Sep 17 00:00:00 2001 From: Andreas Mager Date: Wed, 1 Feb 2017 15:50:39 +0100 Subject: [PATCH] broker/InternalRepublisher: code formatter used --- .../spi/impl/InternalRepublisher.java | 61 +++++++++++++------ 1 file changed, 44 insertions(+), 17 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index 13d3e686..7db968a0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -1,5 +1,5 @@ -package io.moquette.spi.impl; +package io.moquette.spi.impl; import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; @@ -7,7 +7,6 @@ import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.util.ArrayList; import java.util.Collection; import java.util.List; @@ -25,15 +24,20 @@ class InternalRepublisher { void publishRetained(ClientSession targetSession, Collection messages) { for (IMessagesStore.StoredMessage storedMsg : messages) { - //fire as retained the message + // fire as retained the message Integer packetID = storedMsg.getQos() == MqttQoS.AT_MOST_ONCE ? null : targetSession.nextPacketId(); if (packetID != null) { - LOG.debug("Adding message to inflight zone. MqttClientId = {}, packetId = {}, messageId = {}, topic = {}.", - targetSession.clientID, packetID, storedMsg.getMessageID(), storedMsg.getTopic()); + LOG.debug( + "Adding message to inflight zone. " + + "MqttClientId = {}, packetId = {}, messageId = {}, topic = {}.", + targetSession.clientID, + packetID, + storedMsg.getMessageID(), + storedMsg.getTopic()); targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID); } MqttPublishMessage publishMsg = retainedPublish(storedMsg); - //set the PacketIdentifier only for QoS > 0 + // set the PacketIdentifier only for QoS > 0 if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) { publishMsg = retainedPublish(storedMsg, packetID); } @@ -41,19 +45,21 @@ class InternalRepublisher { } } - - void publishStored(ClientSession clientSession, BlockingQueue publishedEvents) { List storedPublishes = new ArrayList<>(); publishedEvents.drainTo(storedPublishes); for (IMessagesStore.StoredMessage pubEvt : storedPublishes) { - //put in flight zone - LOG.debug("Adding message ot inflight zone. MqttClientId = {}, guid = {}, messageId = {}, topic = {}.", - clientSession.clientID, pubEvt.getGuid(), pubEvt.getMessageID(), pubEvt.getTopic()); + // put in flight zone + LOG.debug( + "Adding message ot inflight zone. MqttClientId = {}, guid = {}, messageId = {}, topic = {}.", + clientSession.clientID, + pubEvt.getGuid(), + pubEvt.getMessageID(), + pubEvt.getTopic()); clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID()); MqttPublishMessage publishMsg = notRetainedPublish(pubEvt); - //set the PacketIdentifier only for QoS > 0 + // set the PacketIdentifier only for QoS > 0 if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) { publishMsg = notRetainedPublish(pubEvt, pubEvt.getMessageID()); } @@ -62,22 +68,43 @@ class InternalRepublisher { } private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage, Integer messageID) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, messageID); + return createPublishForQos( + storedMessage.getTopic(), + storedMessage.getQos(), + storedMessage.getMessage(), + false, + messageID); } private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, 0); + return createPublishForQos( + storedMessage.getTopic(), + storedMessage.getQos(), + storedMessage.getMessage(), + false, + 0); } private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, 0); + return createPublishForQos( + storedMessage.getTopic(), + storedMessage.getQos(), + storedMessage.getMessage(), + true, + 0); } private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage, Integer packetID) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, packetID); + return createPublishForQos( + storedMessage.getTopic(), + storedMessage.getQos(), + storedMessage.getMessage(), + true, + packetID); } - public static MqttPublishMessage createPublishForQos(String topic, MqttQoS qos, ByteBuf message, boolean retained, int messageId) { + public static MqttPublishMessage createPublishForQos(String topic, MqttQoS qos, ByteBuf message, boolean retained, + int messageId) { MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0); MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId); return new MqttPublishMessage(fixedHeader, varHeader, message);