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

broker/Qos1PublishHandler: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:39 +01:00
parent a7d476589a
commit b7f3e1beba
1 changed files with 37 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+37 -24
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;
@@ -14,14 +15,13 @@ import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.List; import java.util.List;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
class Qos1PublishHandler extends QosPublishHandler { class Qos1PublishHandler extends QosPublishHandler {
private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class);
private final SubscriptionsStore subscriptions; private final SubscriptionsStore subscriptions;
@@ -31,10 +31,9 @@ class Qos1PublishHandler extends QosPublishHandler {
private final MessagesPublisher publisher; private final MessagesPublisher publisher;
public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions,
IMessagesStore messagesStore, BrokerInterceptor interceptor, IMessagesStore messagesStore, BrokerInterceptor interceptor,
ConnectionDescriptorStore connectionDescriptors, ConnectionDescriptorStore connectionDescriptors, String brokerPort, MessagesPublisher messagesPublisher) {
String brokerPort, MessagesPublisher messagesPublisher) { super(authorizator);
super(authorizator);
this.subscriptions = subscriptions; this.subscriptions = subscriptions;
this.m_messagesStore = messagesStore; this.m_messagesStore = messagesStore;
this.m_interceptor = interceptor; this.m_interceptor = interceptor;
@@ -43,7 +42,7 @@ class Qos1PublishHandler extends QosPublishHandler {
} }
void receivedPublishQos1(Channel channel, MqttPublishMessage msg) { void receivedPublishQos1(Channel channel, MqttPublishMessage msg) {
//verify if topic can be write // verify if topic can be write
final String topic = msg.variableHeader().topicName(); final String topic = msg.variableHeader().topicName();
if (checkWriteOnTopic(topic, channel)) { if (checkWriteOnTopic(topic, channel)) {
return; return;
@@ -51,33 +50,41 @@ class Qos1PublishHandler extends QosPublishHandler {
final int messageID = msg.variableHeader().messageId(); final int messageID = msg.variableHeader().messageId();
//route message to subscribers // route message to subscribers
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel); String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID); toStoreMsg.setClientID(clientID);
if (LOG.isTraceEnabled()) { if (LOG.isTraceEnabled()) {
LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", LOG.trace(
clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), "Sending publish message to subscribers. "
subscriptions.dumpTree()); + "MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.",
} else { clientID,
LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID, topic,
topic, messageID); messageID,
} DebugUtils.payload2Str(toStoreMsg.getMessage()),
subscriptions.dumpTree());
} else {
LOG.info(
"Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.",
clientID,
topic,
messageID);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic); List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions);
//send PUBACK // send PUBACK
//TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) { // TODO Don't send PUBREC for Hz publish notification, if (msg.isLocal()) {
sendPubAck(clientID, messageID); sendPubAck(clientID, messageID);
// } // }
if (msg.fixedHeader().isRetain()) { if (msg.fixedHeader().isRetain()) {
if (!msg.payload().isReadable()) { if (!msg.payload().isReadable()) {
m_messagesStore.cleanRetained(topic); m_messagesStore.cleanRetained(topic);
} else { } else {
//before wasn't stored // before wasn't stored
MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg);
m_messagesStore.storeRetained(topic, guid); m_messagesStore.storeRetained(topic, guid);
} }
@@ -94,14 +101,20 @@ class Qos1PublishHandler extends QosPublishHandler {
try { try {
if (connectionDescriptors == null) { 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); LOG.debug("clientIDs are {}", connectionDescriptors);
if (!connectionDescriptors.isConnected(clientId)) { 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); connectionDescriptors.sendMessage(pubAckMessage, messageID, clientId);
} catch(Throwable t) { } catch (Throwable t) {
LOG.error(null, t); LOG.error(null, t);
} }
} }