From 32244e9482367596b312e3ec944ea6cc17c67d49 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 13 Dec 2016 23:02:54 +0100 Subject: [PATCH] Peeled common code --- .../java/io/moquette/spi/impl/MessagesPublisher.java | 10 +++------- .../spi/impl/PersistentQueueMessageSender.java | 7 +++---- 2 files changed, 6 insertions(+), 11 deletions(-) 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 19524cae..5c5aa4b7 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -56,19 +56,15 @@ class MessagesPublisher { sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); ByteBuffer message = origMessage.duplicate(); if (targetIsActive) { - //QoS 0 - if (qos == AbstractMessage.QOSType.MOST_ONE) { - PublishMessage publishMsg = createPublishForQos(topic, qos, message); - this.persistentSender.publishQos2(targetSession, publishMsg); - } else { + PublishMessage publishMsg = createPublishForQos(topic, qos, message); + if (qos != AbstractMessage.QOSType.MOST_ONE) { //QoS 1 or 2 int messageId = targetSession.nextPacketId(); targetSession.inFlightAckWaiting(guid, messageId); - PublishMessage publishMsg = createPublishForQos(topic, qos, message); //set the PacketIdentifier only for QoS > 0 publishMsg.setMessageID(messageId); - this.persistentSender.publishQos2(targetSession, publishMsg); } + this.persistentSender.sendPublish(targetSession, publishMsg); } else { if (!targetSession.isCleanSession()) { //store the message in targetSession queue to deliver 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 a4fd2fec..90a30ee1 100644 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java @@ -22,13 +22,12 @@ class PersistentQueueMessageSender { this.connectionDescriptors = connectionDescriptors; } - void publishQos2(ClientSession clientsession, PublishMessage pubMessage) { + void sendPublish(ClientSession clientsession, PublishMessage pubMessage) { String clientId = clientsession.clientID; - LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", - clientId, pubMessage.getTopicName(), pubMessage.getQos(), false, pubMessage.getMessageID()); - LOG.info("send publish message to <{}> on topic <{}>", clientId, pubMessage.getTopicName()); if (LOG.isDebugEnabled()) { + LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", + clientId, pubMessage.getTopicName(), pubMessage.getQos(), false, pubMessage.getMessageID()); LOG.debug("content <{}>", DebugUtils.payload2Str(pubMessage.getPayload())); }