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

Minor, unified method

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:02 +01:00
parent 57d9eeef64
commit 2640a5354c
2 changed files with 18 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java
+12 -5
View File
@@ -11,6 +11,8 @@ import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
class BestEffortMessageSender {
private static final Logger LOG = LoggerFactory.getLogger(BestEffortMessageSender.class);
@@ -24,11 +26,7 @@ class BestEffortMessageSender {
String clientId = clientsession.clientID;
LOG.debug("publishQos0 invoked clientId <{}> on topic <{}>", clientId, topic);
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(topic);
pubMessage.setQos(AbstractMessage.QOSType.MOST_ONE);
pubMessage.setPayload(message);
PublishMessage pubMessage = createPublishForQos(topic, MOST_ONE, message);
LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
if (LOG.isDebugEnabled()) {
@@ -55,4 +53,13 @@ class BestEffortMessageSender {
channel.writeAndFlush(pubMessage);
}
}
static PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
return pubMessage;
}
}
broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
+6 -9
View File
@@ -11,6 +11,8 @@ import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.spi.impl.BestEffortMessageSender.createPublishForQos;
class PersistentQueueMessageSender {
private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class);
@@ -25,20 +27,15 @@ class PersistentQueueMessageSender {
String clientId = clientsession.clientID;
LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}",
clientId, topic, qos, false, messageID);
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(false);
pubMessage.setTopicName(topic);
pubMessage.setQos(qos);
pubMessage.setPayload(message);
PublishMessage pubMessage = createPublishForQos(topic, qos, message);
//set the PacketIdentifier only for QoS > 0
pubMessage.setMessageID(messageID);
LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(message));
}
//set the PacketIdentifier only for QoS > 0
if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
pubMessage.setMessageID(messageID);
}
if (connectionDescriptors == null) {
throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " +