From 1c6cbd5d30c522f573ba58ef27595ba130d39e42 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 13 Dec 2016 22:54:59 +0100 Subject: [PATCH] Removed the necessity to use BestEffortMessageSender --- .../spi/impl/BestEffortMessageSender.java | 9 +-------- .../io/moquette/spi/impl/InternalRepublisher.java | 4 +++- .../io/moquette/spi/impl/MessagesPublisher.java | 9 +++------ .../spi/impl/PersistentQueueMessageSender.java | 15 ++++++++++++++- 4 files changed, 21 insertions(+), 16 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java b/broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java index 7e632ea8..795f6dce 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java @@ -48,12 +48,5 @@ class BestEffortMessageSender { } } - 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; - } + } diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index 1eec5548..02d88548 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -15,6 +15,8 @@ import java.util.Collection; import java.util.List; import java.util.concurrent.ConcurrentMap; +import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE; + class InternalRepublisher { private static final Logger LOG = LoggerFactory.getLogger(InternalRepublisher.class); @@ -88,7 +90,7 @@ class InternalRepublisher { LOG.debug("channel is writable"); //if channel is writable don't enqueue channel.writeAndFlush(pubMessage); - } else { + } else if (pubMessage.getQos() != MOST_ONE) { //enqueue to the client session LOG.debug("enqueue to client session"); clientsession.enqueue(pubMessage); diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 26e82557..19524cae 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -15,8 +15,7 @@ import java.nio.ByteBuffer; import java.util.List; import java.util.concurrent.ConcurrentMap; -import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE; -import static io.moquette.spi.impl.BestEffortMessageSender.createPublishForQos; +import static io.moquette.spi.impl.PersistentQueueMessageSender.createPublishForQos; import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; class MessagesPublisher { @@ -25,7 +24,6 @@ class MessagesPublisher { private final ConcurrentMap connectionDescriptors; private final ISessionsStore m_sessionsStore; private final IMessagesStore m_messagesStore; - private final BestEffortMessageSender bestEffortSender; private final PersistentQueueMessageSender persistentSender; public MessagesPublisher(ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, @@ -33,7 +31,6 @@ class MessagesPublisher { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; this.m_messagesStore = messagesStore; - this.bestEffortSender = new BestEffortMessageSender(connectionDescriptors); this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors); } @@ -61,8 +58,8 @@ class MessagesPublisher { if (targetIsActive) { //QoS 0 if (qos == AbstractMessage.QOSType.MOST_ONE) { - PublishMessage publishMsg = BestEffortMessageSender.createPublishForQos(topic, MOST_ONE, message); - this.bestEffortSender.publishQos0(targetSession, publishMsg); + PublishMessage publishMsg = createPublishForQos(topic, qos, message); + this.persistentSender.publishQos2(targetSession, publishMsg); } else { //QoS 1 or 2 int messageId = targetSession.nextPacketId(); diff --git a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java index 5aca1047..a4fd2fec 100644 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java @@ -1,5 +1,6 @@ package io.moquette.spi.impl; +import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.parser.proto.messages.PublishMessage; import io.moquette.server.ConnectionDescriptor; import io.moquette.spi.ClientSession; @@ -7,8 +8,11 @@ import io.netty.channel.Channel; import org.slf4j.Logger; 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 PersistentQueueMessageSender { private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class); @@ -44,11 +48,20 @@ class PersistentQueueMessageSender { LOG.debug("channel is writable"); //if channel is writable don't enqueue channel.writeAndFlush(pubMessage); - } else { + } else if (pubMessage.getQos() != MOST_ONE) { //enqueue to the client session LOG.debug("enqueue to client session"); clientsession.enqueue(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; + } + }