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

broker/MessagesPublisher: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:39 +01:00
parent 91ca7fd1c2
commit a308780acd
1 changed files with 24 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+24 -15
View File
@@ -1,3 +1,4 @@
package io.moquette.spi.impl;
import io.moquette.server.ConnectionDescriptorStore;
@@ -10,9 +11,7 @@ import io.netty.buffer.ByteBuf;
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.lowerQosToTheSubscriptionDesired;
class MessagesPublisher {
@@ -24,7 +23,7 @@ class MessagesPublisher {
private final PersistentQueueMessageSender messageSender;
public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) {
IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) {
this.connectionDescriptors = connectionDescriptors;
this.m_sessionsStore = sessionsStore;
this.m_messagesStore = messagesStore;
@@ -35,7 +34,8 @@ class MessagesPublisher {
return notRetainedPublishWithMessageId(topic, qos, message, 0);
}
private static MqttPublishMessage notRetainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message, int messageId) {
private static MqttPublishMessage notRetainedPublishWithMessageId(String topic, MqttQoS qos, ByteBuf message,
int messageId) {
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(topic, messageId);
return new MqttPublishMessage(fixedHeader, varHeader, message);
@@ -46,38 +46,47 @@ class MessagesPublisher {
final MqttQoS publishingQos = pubMsg.getQos();
final ByteBuf origMessage = pubMsg.getMessage();
//if QoS 1 or 2 store the message
//TODO perhaps this block is not needed.
// if QoS 1 or 2 store the message
// TODO perhaps this block is not needed.
MessageGUID guid = null;
if (publishingQos != MqttQoS.AT_MOST_ONCE) {
guid = m_messagesStore.storePublishForFuture(pubMsg);
}
for (final Subscription sub : topicMatchingSubscriptions) {
MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos);
ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId());
boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId());
//we need to retain because duplicate only copy r/w indexes and don't retain() causing refCnt = 0
// we need to retain because duplicate only copy r/w indexes and don't retain() causing
// refCnt = 0
ByteBuf message = origMessage.retainedDuplicate();
if (targetIsActive) {
LOG.debug("Sending PUBLISH message to active subscriber. MessageId={}, CId={}, topicFilter={}, qos={}",
pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos);
LOG.debug(
"Sending PUBLISH message to active subscriber. MessageId={}, CId={}, topicFilter={}, qos={}",
pubMsg.getMessageID(),
sub.getClientId(),
sub.getTopicFilter(),
qos);
MqttPublishMessage publishMsg = notRetainedPublish(topic, qos, message);
if (qos != MqttQoS.AT_MOST_ONCE) {
//QoS 1 or 2
// QoS 1 or 2
int messageId = targetSession.nextPacketId();
targetSession.inFlightAckWaiting(guid, messageId);
//set the PacketIdentifier only for QoS > 0
// set the PacketIdentifier only for QoS > 0
publishMsg = notRetainedPublishWithMessageId(topic, qos, message, messageId);
}
this.messageSender.sendPublish(targetSession, publishMsg);
} else {
if (!targetSession.isCleanSession()) {
LOG.debug("Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}",
pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos);
//store the message in targetSession queue to deliver
LOG.debug(
"Storing pending PUBLISH inactive message. MessageId={}, CId={}, topicFilter={}, qos={}",
pubMsg.getMessageID(),
sub.getClientId(),
sub.getTopicFilter(),
qos);
// store the message in targetSession queue to deliver
targetSession.enqueue(pubMsg);
}
}