Assure intercept methods to not change the message

This was done by cloning.

broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java

@@ -35,7 +35,7 @@ final class BrokerInterceptor implements Interceptor {
 
     @Override
     public void notifyClientConnected(final ConnectMessage msg) {
-        handler.onConnect(msg);
+        handler.onConnect(msg.readOnlyClone());
     }
 
     @Override
@@ -45,16 +45,16 @@ final class BrokerInterceptor implements Interceptor {
 
     @Override
     public void notifyTopicPublished(final PublishMessage msg) {
-        handler.onPublish(msg);
+        handler.onPublish(msg.readOnlyClone());
     }
 
     @Override
     public void notifyTopicSubscribed(final Subscription sub) {
-        handler.onSubscribe(sub);
+        handler.onSubscribe(sub.clone());
     }
 
     @Override
     public void notifyTopicUnsubscribed(final Subscription sub) {
-        handler.onUnsubscribe(sub);
+        handler.onUnsubscribe(sub.clone());
     }
 }

broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Subscription.java

@@ -106,4 +106,13 @@ public class Subscription implements Serializable {
     public String toString() {
         return String.format("[filter:%s, cliID: %s, qos: %s, active: %s]", this.topicFilter, this.clientId, this.requestedQos, this.active);
     }
+
+    @Override
+    public Subscription clone() {
+        try {
+            return (Subscription) super.clone();
+        } catch (CloneNotSupportedException e) {
+            return null;
+        }
+    }
 }

parser_commons/src/main/java/org/eclipse/moquette/proto/messages/ConnectMessage.java

@@ -164,4 +164,12 @@ public class ConnectMessage extends AbstractMessage {
         }
         return base;
     }
+
+    public ConnectMessage readOnlyClone() {
+        try {
+            return (ConnectMessage) super.clone();
+        } catch (CloneNotSupportedException e) {
+            return null;
+        }
+    }
 }

parser_commons/src/main/java/org/eclipse/moquette/proto/messages/PublishMessage.java

@@ -54,4 +54,14 @@ public class PublishMessage extends MessageIDMessage {
     public void setPayload(ByteBuffer payload) {
         this.m_payload = payload;
     }
+
+    public PublishMessage readOnlyClone() {
+        try {
+            final PublishMessage clone = (PublishMessage) super.clone();
+            clone.m_payload = this.m_payload.asReadOnlyBuffer();
+            return clone;
+        } catch (CloneNotSupportedException e) {
+            return null;
+        }
+    }
 }