anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

broker/InternalRepublisher: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:39 +01:00
parent 0655f20543
commit 91ca7fd1c2
1 changed files with 44 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+44 -17
View File
@@ -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<IMessagesStore.StoredMessage> 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<IMessagesStore.StoredMessage> publishedEvents) {
List<IMessagesStore.StoredMessage> 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);