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

broker/PersistentQueueMessageSender: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:39 +01:00
parent a308780acd
commit 6eb635f437
1 changed files with 32 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
+32 -18
View File
@@ -1,3 +1,4 @@
package io.moquette.spi.impl; package io.moquette.spi.impl;
import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.ConnectionDescriptorStore;
@@ -6,7 +7,6 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
@@ -21,27 +21,41 @@ class PersistentQueueMessageSender {
void sendPublish(ClientSession clientsession, MqttPublishMessage pubMessage) { void sendPublish(ClientSession clientsession, MqttPublishMessage pubMessage) {
String clientId = clientsession.clientID; String clientId = clientsession.clientID;
final int messageId = pubMessage.variableHeader().messageId(); final int messageId = pubMessage.variableHeader().messageId();
final String topicName = pubMessage.variableHeader().topicName(); final String topicName = pubMessage.variableHeader().topicName();
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId, LOG.debug(
clientId, topicName, DebugUtils.payload2Str(pubMessage.payload())); "Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}",
messageId,
clientId,
topicName,
DebugUtils.payload2Str(pubMessage.payload()));
} else { } else {
LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName); LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName);
} }
boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, messageId, clientId); boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, messageId, clientId);
MqttQoS qos = pubMessage.fixedHeader().qosLevel(); MqttQoS qos = pubMessage.fixedHeader().qosLevel();
if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) { if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) {
LOG.warn( LOG.warn(
"The PUBLISH message could not be delivered. It will be stored. MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.", "The PUBLISH message could not be delivered. It will be stored. "
messageId, clientId, topicName, qos, false); + "MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.",
clientsession.enqueue(asStoredMessage(pubMessage)); messageId,
} else { clientId,
LOG.warn( topicName,
"The PUBLISH message could not be delivered. It will be discarded. MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.", qos,
messageId, clientId, topicName, qos, true); false);
} clientsession.enqueue(asStoredMessage(pubMessage));
} else {
LOG.warn(
"The PUBLISH message could not be delivered. It will be discarded. "
+ "MessageId = {}, CId={}, topic = {}, qos = {}, cleanSession = {}.",
messageId,
clientId,
topicName,
qos,
true);
}
} }
} }