From 968e02171179a9f207d046c028d95eef406b5378 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 12 Dec 2016 23:42:40 +0100 Subject: [PATCH] Reintroduced bad removed conditions --- .../io/moquette/spi/impl/Qos1Publisher.java | 48 ++++++++++++------- 1 file changed, 30 insertions(+), 18 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java b/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java index f1cc270f..e0a9c07f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java @@ -50,39 +50,51 @@ class Qos1Publisher { LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); ByteBuffer message = origMessage.duplicate(); - //QoS 1 or 2 - //if the target subscription is not clean session and is not connected => store it - if (!targetSession.isCleanSession() && !targetIsActive) { - //store the message in targetSession queue to deliver - targetSession.enqueueToDeliver(guid); - } else { - //publish - if (targetIsActive) { - int messageId = targetSession.nextPacketId(); - targetSession.inFlightAckWaiting(guid, messageId); - publishQos1(targetSession, topic, message, messageId, qos); + if (qos == AbstractMessage.QOSType.MOST_ONE && targetIsActive) { + //QoS 0 + publishQos1(targetSession, topic, qos, message, false, null); + } else { + //QoS 1 or 2 + //if the target subscription is not clean session and is not connected => store it + if (!targetSession.isCleanSession() && !targetIsActive) { + //store the message in targetSession queue to deliver + targetSession.enqueueToDeliver(guid); + } else { + //publish + if (targetIsActive) { + int messageId = targetSession.nextPacketId(); + targetSession.inFlightAckWaiting(guid, messageId); + publishQos1(targetSession, topic, qos, message, false, messageId); + } } } } } - private void publishQos1(ClientSession clientsession, String topic, - ByteBuffer message, int messageID, AbstractMessage.QOSType qos) { + private void publishQos1(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, + ByteBuffer message, boolean retained, Integer messageID) { String clientId = clientsession.clientID; - LOG.debug("publishQos1 invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", - clientId, topic, qos, false, messageID); + LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", + clientId, topic, qos, retained, messageID); PublishMessage pubMessage = new PublishMessage(); - pubMessage.setRetainFlag(false); + pubMessage.setRetainFlag(retained); pubMessage.setTopicName(topic); pubMessage.setQos(qos); pubMessage.setPayload(message); - //QoS > 0 => set the PacketIdentifier - 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); + } else { + if (messageID != null) { + throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message " + + "with PacketIdentifier: " + messageID); + } + } if (connectionDescriptors == null) { throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " +