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

broker/Qos2PublishHandler: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:40 +01:00
parent b7f3e1beba
commit 127894b2f9
1 changed files with 49 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+49 -36
View File
@@ -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<Subscription> 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) {