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 aa176742..f70c08aa 100644
--- a/broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java
+++ b/broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java
@@ -11,8 +11,6 @@ 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);
@@ -22,12 +20,8 @@ class BestEffortMessageSender {
         this.connectionDescriptors = connectionDescriptors;
     }
 
-    void publishQos0(ClientSession clientsession, String topic, ByteBuffer message) {
+    void publishQos0(ClientSession clientsession, String topic, ByteBuffer message, PublishMessage pubMessage) {
         String clientId = clientsession.clientID;
-
-        LOG.debug("publishQos0 invoked clientId <{}> on topic <{}>", clientId, topic);
-        PublishMessage pubMessage = createPublishForQos(topic, MOST_ONE, message);
-
         LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
         if (LOG.isDebugEnabled()) {
             LOG.debug("content <{}>", DebugUtils.payload2Str(message));
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 faec8218..f04c48c9 100644
--- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
@@ -1,6 +1,7 @@
 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;
 import io.moquette.spi.IMessagesStore;
@@ -14,6 +15,8 @@ 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.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
 
 class MessagesPublisher {
@@ -58,12 +61,16 @@ class MessagesPublisher {
             if (targetIsActive) {
                 //QoS 0
                 if (qos == AbstractMessage.QOSType.MOST_ONE) {
-                    this.bestEffortSender.publishQos0(targetSession, topic, message);
+                    PublishMessage publishMsg = BestEffortMessageSender.createPublishForQos(topic, MOST_ONE, message);
+                    this.bestEffortSender.publishQos0(targetSession, topic, message, publishMsg);
                 } else {
                     //QoS 1 or 2
                     int messageId = targetSession.nextPacketId();
                     targetSession.inFlightAckWaiting(guid, messageId);
-                    this.persistentSender.publishQos2(targetSession, topic, qos, message, messageId);
+                    PublishMessage publishMsg = createPublishForQos(topic, qos, message);
+                    //set the PacketIdentifier only for QoS > 0
+                    publishMsg.setMessageID(messageId);
+                    this.persistentSender.publishQos2(targetSession, publishMsg);
                 }
             } else {
                 if (!targetSession.isCleanSession()) {
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 1fcdd197..640d3f6b 100644
--- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
+++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
@@ -11,8 +11,6 @@ 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);
@@ -22,19 +20,14 @@ class PersistentQueueMessageSender {
         this.connectionDescriptors = connectionDescriptors;
     }
 
-    void publishQos2(ClientSession clientsession, String topic, AbstractMessage.QOSType qos,
-                             ByteBuffer message, int messageID) {
+    void publishQos2(ClientSession clientsession, PublishMessage pubMessage) {
         String clientId = clientsession.clientID;
         LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}",
-                clientId, topic, qos, false, messageID);
-        PublishMessage pubMessage = createPublishForQos(topic, qos, message);
+                clientId, pubMessage.getTopicName(), pubMessage.getQos(), false, pubMessage.getMessageID());
 
-        //set the PacketIdentifier only for QoS > 0
-        pubMessage.setMessageID(messageID);
-
-        LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
+        LOG.info("send publish message to <{}> on topic <{}>", clientId, pubMessage.getTopicName());
         if (LOG.isDebugEnabled()) {
-            LOG.debug("content <{}>", DebugUtils.payload2Str(message));
+            LOG.debug("content <{}>", DebugUtils.payload2Str(pubMessage.getPayload()));
         }
 
         if (connectionDescriptors == null) {