diff --git a/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java b/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java index ef0bd0ca..a66aef8d 100644 --- a/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java +++ b/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); } diff --git a/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java b/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java index a56542fc..957f5046 100644 --- a/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java +++ b/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); } diff --git a/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptAbstractMessage.java b/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptAbstractMessage.java new file mode 100644 index 00000000..f8720eaf --- /dev/null +++ b/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(); + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptConnectMessage.java b/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptConnectMessage.java new file mode 100644 index 00000000..4766f744 --- /dev/null +++ b/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(); + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptDisconnectMessage.java b/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptDisconnectMessage.java new file mode 100644 index 00000000..844c4059 --- /dev/null +++ b/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; + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptPublishMessage.java b/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptPublishMessage.java new file mode 100644 index 00000000..21d3f26b --- /dev/null +++ b/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; + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptSubscribeMessage.java b/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptSubscribeMessage.java new file mode 100644 index 00000000..2e82fd8f --- /dev/null +++ b/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(); + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptUnsubscribeMessage.java b/broker/src/main/java/org/eclipse/moquette/interception/messages/InterceptUnsubscribeMessage.java new file mode 100644 index 00000000..8fcbf003 --- /dev/null +++ b/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; + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java index 5a632342..af39c2a1 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java +++ b/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)); } }); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index fc111731..a7077088 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -306,7 +306,7 @@ class ProtocolProcessor implements EventHandler { 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 { subscriptions.removeSubscription(topic, clientID); m_sessionsStore.removeSubscription(topic, clientID); - m_interceptor.notifyTopicUnsubscribed(topic); + m_interceptor.notifyTopicUnsubscribed(topic, clientID); } //ack the client diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiableConnectMessage.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiableConnectMessage.java deleted file mode 100644 index 07fe6758..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiableConnectMessage.java +++ /dev/null @@ -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(); - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiablePublishMessage.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiablePublishMessage.java deleted file mode 100644 index aaa11d76..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiablePublishMessage.java +++ /dev/null @@ -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(); - } -} diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java index c417bef7..56e33957 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java +++ b/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()); }