Encapsulating every intercept data

- Removed UnmodifiableConnectMessage and UnmodifiablePublishMessage
  classes.

broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java

@@ -15,6 +15,11 @@
  */
 package org.eclipse.moquette.interception;
 
+import org.eclipse.moquette.interception.messages.InterceptConnectMessage;
+import org.eclipse.moquette.interception.messages.InterceptDisconnectMessage;
+import org.eclipse.moquette.interception.messages.InterceptPublishMessage;
+import org.eclipse.moquette.interception.messages.InterceptSubscribeMessage;
+import org.eclipse.moquette.interception.messages.InterceptUnsubscribeMessage;
 import org.eclipse.moquette.proto.messages.*;
 import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
 
@@ -32,13 +37,13 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
  */
 public interface InterceptHandler {
 
-    void onConnect(ConnectMessage msg);
+    void onConnect(InterceptConnectMessage msg);
 
-    void onDisconnect(String clientID);
+    void onDisconnect(InterceptDisconnectMessage msg);
 
-    void onPublish(PublishMessage msg);
+    void onPublish(InterceptPublishMessage msg);
 
-    void onSubscribe(Subscription sub);
+    void onSubscribe(InterceptSubscribeMessage msg);
 
-    void onUnsubscribe(String topic);
+    void onUnsubscribe(InterceptUnsubscribeMessage msg);
 }

broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java

@@ -38,10 +38,10 @@ public interface Interceptor {
 
     void notifyClientDisconnected(String clientID);
 
-    void notifyTopicPublished(PublishMessage msg);
+    void notifyTopicPublished(PublishMessage msg, String clientID);
 
     void notifyTopicSubscribed(Subscription sub);
 
-    void notifyTopicUnsubscribed(String topic);
+    void notifyTopicUnsubscribed(String topic, String clientID);
 
 }

broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptAbstractMessage.java

@@ -0,0 +1,26 @@
+package org.eclipse.moquette.interception.messages;
+
+import org.eclipse.moquette.proto.messages.AbstractMessage;
+
+/**
+ * @author Wagner Macedo
+ */
+public abstract class InterceptAbstractMessage {
+    private final AbstractMessage msg;
+
+    InterceptAbstractMessage(AbstractMessage msg) {
+        this.msg = msg;
+    }
+
+    public boolean isRetainFlag() {
+        return msg.isRetainFlag();
+    }
+
+    public boolean isDupFlag() {
+        return msg.isDupFlag();
+    }
+
+    public AbstractMessage.QOSType getQos() {
+        return msg.getQos();
+    }
+}

broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptConnectMessage.java

@@ -0,0 +1,71 @@
+package org.eclipse.moquette.interception.messages;
+
+import org.eclipse.moquette.proto.messages.ConnectMessage;
+
+/**
+ * @author Wagner Macedo
+ */
+public class InterceptConnectMessage extends InterceptAbstractMessage {
+    private final ConnectMessage msg;
+
+    public InterceptConnectMessage(ConnectMessage msg) {
+        super(msg);
+        this.msg = msg;
+    }
+
+    public String getClientID() {
+        return msg.getClientID();
+    }
+
+    public boolean isCleanSession() {
+        return msg.isCleanSession();
+    }
+
+    public int getKeepAlive() {
+        return msg.getKeepAlive();
+    }
+
+    public boolean isPasswordFlag() {
+        return msg.isPasswordFlag();
+    }
+
+    public byte getProtocolVersion() {
+        return msg.getProtocolVersion();
+    }
+
+    public String getProtocolName() {
+        return msg.getProtocolName();
+    }
+
+    public boolean isUserFlag() {
+        return msg.isUserFlag();
+    }
+
+    public boolean isWillFlag() {
+        return msg.isWillFlag();
+    }
+
+    public byte getWillQos() {
+        return msg.getWillQos();
+    }
+
+    public boolean isWillRetain() {
+        return msg.isWillRetain();
+    }
+
+    public String getUsername() {
+        return msg.getUsername();
+    }
+
+    public String getPassword() {
+        return msg.getPassword();
+    }
+
+    public String getWillTopic() {
+        return msg.getWillTopic();
+    }
+
+    public String getWillMessage() {
+        return msg.getWillMessage();
+    }
+}

broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptDisconnectMessage.java

@@ -0,0 +1,16 @@
+package org.eclipse.moquette.interception.messages;
+
+/**
+ * @author Wagner Macedo
+ */
+public class InterceptDisconnectMessage {
+    private final String clientID;
+
+    public InterceptDisconnectMessage(String clientID) {
+        this.clientID = clientID;
+    }
+
+    public String getClientID() {
+        return clientID;
+    }
+}

broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptPublishMessage.java

@@ -0,0 +1,31 @@
+package org.eclipse.moquette.interception.messages;
+
+import org.eclipse.moquette.proto.messages.PublishMessage;
+
+import java.nio.ByteBuffer;
+
+/**
+ * @author Wagner Macedo
+ */
+public class InterceptPublishMessage extends InterceptAbstractMessage {
+    private final PublishMessage msg;
+    private final String clientID;
+
+    public InterceptPublishMessage(PublishMessage msg, String clientID) {
+        super(msg);
+        this.msg = msg;
+        this.clientID = clientID;
+    }
+
+    public String getTopicName() {
+        return msg.getTopicName();
+    }
+
+    public ByteBuffer getPayload() {
+        return msg.getPayload();
+    }
+
+    public String getClientID() {
+        return clientID;
+    }
+}

broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptSubscribeMessage.java

@@ -0,0 +1,27 @@
+package org.eclipse.moquette.interception.messages;
+
+import org.eclipse.moquette.proto.messages.AbstractMessage;
+import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
+
+/**
+ * @author Wagner Macedo
+ */
+public class InterceptSubscribeMessage {
+    private final Subscription subscription;
+
+    public InterceptSubscribeMessage(Subscription subscription) {
+        this.subscription = subscription;
+    }
+
+    public String getClientID() {
+        return subscription.getClientId();
+    }
+
+    public AbstractMessage.QOSType getRequestedQos() {
+        return subscription.getRequestedQos();
+    }
+
+    public String getTopicFilter() {
+        return subscription.getTopicFilter();
+    }
+}

broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptUnsubscribeMessage.java

@@ -0,0 +1,22 @@
+package org.eclipse.moquette.interception.messages;
+
+/**
+ * @author Wagner Macedo
+ */
+public class InterceptUnsubscribeMessage {
+    private final String topicFilter;
+    private final String clientID;
+
+    public InterceptUnsubscribeMessage(String topicFilter, String clientID) {
+        this.topicFilter = topicFilter;
+        this.clientID = clientID;
+    }
+
+    public String getTopicFilter() {
+        return topicFilter;
+    }
+
+    public String getClientID() {
+        return clientID;
+    }
+}

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

@@ -17,6 +17,7 @@ package org.eclipse.moquette.spi.impl;
 
 import org.eclipse.moquette.interception.InterceptHandler;
 import org.eclipse.moquette.interception.Interceptor;
+import org.eclipse.moquette.interception.messages.*;
 import org.eclipse.moquette.proto.messages.ConnectMessage;
 import org.eclipse.moquette.proto.messages.PublishMessage;
 import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
@@ -48,12 +49,11 @@ final class BrokerInterceptor implements Interceptor {
 
     @Override
     public void notifyClientConnected(final ConnectMessage msg) {
-        final ConnectMessage copied = new UnmodifiableConnectMessage(msg);
         for (final InterceptHandler handler : this.handlers) {
             executor.execute(new Runnable() {
                 @Override
                 public void run() {
-                    handler.onConnect(copied);
+                    handler.onConnect(new InterceptConnectMessage(msg));
                 }
             });
         }
@@ -65,20 +65,19 @@ final class BrokerInterceptor implements Interceptor {
             executor.execute(new Runnable() {
                 @Override
                 public void run() {
-                    handler.onDisconnect(clientID);
+                    handler.onDisconnect(new InterceptDisconnectMessage(clientID));
                 }
             });
         }
     }
 
     @Override
-    public void notifyTopicPublished(final PublishMessage msg) {
-        final PublishMessage copied = new UnmodifiablePublishMessage(msg);
+    public void notifyTopicPublished(final PublishMessage msg, final String clientID) {
         for (final InterceptHandler handler : this.handlers) {
             executor.execute(new Runnable() {
                 @Override
                 public void run() {
-                    handler.onPublish(copied);
+                    handler.onPublish(new InterceptPublishMessage(msg, clientID));
                 }
             });
         }
@@ -90,19 +89,19 @@ final class BrokerInterceptor implements Interceptor {
             executor.execute(new Runnable() {
                 @Override
                 public void run() {
-                    handler.onSubscribe(sub);
+                    handler.onSubscribe(new InterceptSubscribeMessage(sub));
                 }
             });
         }
     }
 
     @Override
-    public void notifyTopicUnsubscribed(final String sub) {
+    public void notifyTopicUnsubscribed(final String topic, final String clientID) {
         for (final InterceptHandler handler : this.handlers) {
             executor.execute(new Runnable() {
                 @Override
                 public void run() {
-                    handler.onUnsubscribe(sub);
+                    handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID));
                 }
             });
         }

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

@@ -306,7 +306,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME);
         if (m_authorizator.canWrite(topic, user, clientID)) {
             executePublish(clientID, msg);
-            m_interceptor.notifyTopicPublished(msg);
+            m_interceptor.notifyTopicPublished(msg, clientID);
         } else {
             LOG.debug("topic {} doesn't have write credentials", topic);
         }
@@ -607,7 +607,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
 
             subscriptions.removeSubscription(topic, clientID);
             m_sessionsStore.removeSubscription(topic, clientID);
-            m_interceptor.notifyTopicUnsubscribed(topic);
+            m_interceptor.notifyTopicUnsubscribed(topic, clientID);
         }
 
         //ack the client

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

@@ -1,116 +0,0 @@
-/*
- * Copyright (c) 2012-2015 The original author or authors
- * ------------------------------------------------------
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
- *
- * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
- *
- * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
- *
- * You may elect to redistribute this code under either of these licenses.
- */
-package org.eclipse.moquette.spi.impl;
-
-import org.eclipse.moquette.proto.messages.ConnectMessage;
-
-/**
- * A ConnectMessage that is unmodifiable with copy through copy constructor.
- *
- * @author andrea
- */
-final class UnmodifiableConnectMessage extends ConnectMessage {
-
-    /**
-     * Copy constructor
-     */
-    UnmodifiableConnectMessage(ConnectMessage orig) {
-        this.m_protocolName = orig.getProtocolName();
-        this.m_protocolVersion = orig.getProtocolVersion();
-        this.m_cleanSession = orig.isCleanSession();
-        this.m_willFlag = orig.isWillFlag();
-        this.m_willQos = orig.getWillQos();
-        this.m_willRetain = orig.isWillRetain();
-        this.m_passwordFlag = orig.isPasswordFlag();
-        this.m_userFlag = orig.isUserFlag();
-        this.m_keepAlive = orig.getKeepAlive();
-        this.m_username = orig.getUsername();
-        this.m_password = orig.getPassword();
-        this.m_clientID = orig.getClientID();
-        this.m_willtopic = orig.getWillTopic();
-        this.m_willMessage = orig.getWillMessage();
-    }
-
-    @Override
-    public void setCleanSession(boolean cleanSession) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setKeepAlive(int keepAlive) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setPasswordFlag(boolean passwordFlag) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setProtocolVersion(byte protocolVersion) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setProtocolName(String protocolName) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setUserFlag(boolean userFlag) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setWillFlag(boolean willFlag) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setWillQos(byte willQos) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setWillRetain(boolean willRetain) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setPassword(String password) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setUsername(String username) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setClientID(String clientID) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setWillTopic(String topic) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setWillMessage(String willMessage) {
-        throw new UnsupportedOperationException();
-    }
-}

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

@@ -1,59 +0,0 @@
-/*
- * Copyright (c) 2012-2015 The original author or authors
- * ------------------------------------------------------
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
- *
- * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
- *
- * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
- *
- * You may elect to redistribute this code under either of these licenses.
- */
-package org.eclipse.moquette.spi.impl;
-
-import org.eclipse.moquette.proto.messages.PublishMessage;
-
-import java.nio.ByteBuffer;
-
-
-/**
- * A PublishMessage that is unmodifiable with copy through copy constructor.
- *
- * @author andrea
- */
-final class UnmodifiablePublishMessage extends PublishMessage {
-
-    protected byte m_messageType;
-
-    UnmodifiablePublishMessage(final PublishMessage orig) {
-        this.m_messageID = orig.getMessageID();
-        this.m_topicName = orig.getTopicName();
-        if (orig.getPayload() != null) {
-            this.m_payload = orig.getPayload().asReadOnlyBuffer();
-        }
-        this.m_dupFlag = orig.isDupFlag();
-        this.m_qos = orig.getQos();
-        this.m_retainFlag = orig.isRetainFlag();
-        this.m_remainingLength = orig.getRemainingLength();
-        this.m_messageType = orig.getMessageType();
-    }
-
-    @Override
-    public void setTopicName(String topicName) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setPayload(ByteBuffer payload) {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public void setMessageID(Integer messageID) {
-        throw new UnsupportedOperationException();
-    }
-}

broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java

@@ -15,6 +15,7 @@
  */
 package org.eclipse.moquette.spi.impl;
 
+import org.eclipse.moquette.interception.messages.*;
 import org.eclipse.moquette.interception.InterceptHandler;
 import org.eclipse.moquette.proto.messages.AbstractMessage;
 import org.eclipse.moquette.proto.messages.ConnectMessage;
@@ -39,27 +40,27 @@ public class BrokerInterceptorTest {
     // Interceptor loaded with a custom InterceptHandler special for the tests
     private static final class MockObserver implements InterceptHandler {
         @Override
-        public void onConnect(ConnectMessage msg) {
+        public void onConnect(InterceptConnectMessage msg) {
             n.set(40);
         }
 
         @Override
-        public void onDisconnect(String clientID) {
+        public void onDisconnect(InterceptDisconnectMessage msg) {
             n.set(50);
         }
 
         @Override
-        public void onPublish(PublishMessage msg) {
+        public void onPublish(InterceptPublishMessage msg) {
             n.set(60);
         }
 
         @Override
-        public void onSubscribe(Subscription sub) {
+        public void onSubscribe(InterceptSubscribeMessage msg) {
             n.set(70);
         }
 
         @Override
-        public void onUnsubscribe(String topic) {
+        public void onUnsubscribe(InterceptUnsubscribeMessage msg) {
             n.set(80);
         }
     }
@@ -98,7 +99,7 @@ public class BrokerInterceptorTest {
 
     @Test
     public void testNotifyTopicPublished() throws Exception {
-        interceptor.notifyTopicPublished(new PublishMessage());
+        interceptor.notifyTopicPublished(new PublishMessage(), "cli1234");
         interval();
         assertEquals(60, n.get());
     }
@@ -112,7 +113,7 @@ public class BrokerInterceptorTest {
 
     @Test
     public void testNotifyTopicUnsubscribed() throws Exception {
-        interceptor.notifyTopicUnsubscribed("o2");
+        interceptor.notifyTopicUnsubscribed("o2", "cli1234");
         interval();
         assertEquals(80, n.get());
     }