diff --git a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java index e39e0816..b64c03e6 100644 --- a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java @@ -40,5 +40,5 @@ public abstract class AbstractInterceptHandler implements InterceptHandler { public void onUnsubscribe(InterceptUnsubscribeMessage msg) {} @Override - public void onMessageConsumed(InterceptConsumedMessage msg) {} + public void onMessageAcknowledged(InterceptAcknowledgedMessage msg) {} } diff --git a/broker/src/main/java/io/moquette/interception/InterceptHandler.java b/broker/src/main/java/io/moquette/interception/InterceptHandler.java index 434993e4..c8e905e6 100644 --- a/broker/src/main/java/io/moquette/interception/InterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/InterceptHandler.java @@ -18,7 +18,7 @@ package io.moquette.interception; import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.interception.messages.InterceptConnectMessage; -import io.moquette.interception.messages.InterceptConsumedMessage; +import io.moquette.interception.messages.InterceptAcknowledgedMessage; import io.moquette.interception.messages.InterceptDisconnectMessage; import io.moquette.interception.messages.InterceptPublishMessage; import io.moquette.interception.messages.InterceptSubscribeMessage; @@ -48,5 +48,5 @@ public interface InterceptHandler { void onUnsubscribe(InterceptUnsubscribeMessage msg); - void onMessageConsumed(InterceptConsumedMessage msg); + void onMessageAcknowledged(InterceptAcknowledgedMessage msg); } diff --git a/broker/src/main/java/io/moquette/interception/Interceptor.java b/broker/src/main/java/io/moquette/interception/Interceptor.java index ff73e6a6..be7e4ed1 100644 --- a/broker/src/main/java/io/moquette/interception/Interceptor.java +++ b/broker/src/main/java/io/moquette/interception/Interceptor.java @@ -15,7 +15,7 @@ */ package io.moquette.interception; -import io.moquette.interception.messages.InterceptConsumedMessage; +import io.moquette.interception.messages.InterceptAcknowledgedMessage; import io.moquette.parser.proto.messages.ConnectMessage; import io.moquette.parser.proto.messages.PublishMessage; import io.moquette.spi.impl.subscriptions.Subscription; @@ -45,5 +45,5 @@ public interface Interceptor { void notifyTopicUnsubscribed(String topic, String clientID, final String username); - void notifyMessageConsumed(InterceptConsumedMessage msg); + void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg); } diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java similarity index 70% rename from broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java rename to broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java index 0a01c9bd..6cf88052 100644 --- a/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java @@ -2,12 +2,12 @@ package io.moquette.interception.messages; import static io.moquette.spi.IMessagesStore.StoredMessage; -public class InterceptConsumedMessage { +public class InterceptAcknowledgedMessage { final private StoredMessage msg; private final String username; private final String topic; - public InterceptConsumedMessage( final StoredMessage msg, final String topic, final String username) { + public InterceptAcknowledgedMessage( final StoredMessage msg, final String topic, final String username) { this.msg = msg; this.username = username; this.topic = topic; diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 27732fe8..a7a42cd1 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -108,12 +108,12 @@ final class BrokerInterceptor implements Interceptor { } @Override - public void notifyMessageConsumed( final InterceptConsumedMessage msg ) { + public void notifyMessageAcknowledged( final InterceptAcknowledgedMessage msg ) { for (final InterceptHandler handler : this.handlers) { executor.execute(new Runnable() { @Override public void run() { - handler.onMessageConsumed(msg); + handler.onMessageAcknowledged(msg); } }); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index af14bc3a..287b66ba 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -34,7 +34,7 @@ import io.moquette.spi.impl.subscriptions.Subscription; import static io.moquette.parser.netty.Utils.VERSION_3_1; import static io.moquette.parser.netty.Utils.VERSION_3_1_1; -import io.moquette.interception.messages.InterceptConsumedMessage; +import io.moquette.interception.messages.InterceptAcknowledgedMessage; import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.parser.proto.messages.AbstractMessage.QOSType; import io.moquette.parser.proto.messages.ConnAckMessage; @@ -290,7 +290,7 @@ public class ProtocolProcessor { String topic = inflightMsg.getTopic(); - m_interceptor.notifyMessageConsumed(new InterceptConsumedMessage(inflightMsg, topic, username)); + m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); } private void verifyToActivate(String clientID, ClientSession targetSession) { @@ -590,7 +590,7 @@ public class ProtocolProcessor { targetSession.secondPhaseAcknowledged(messageID); String username = NettyUtils.userName(channel); String topic = inflightMsg.getTopic(); - m_interceptor.notifyMessageConsumed( new InterceptConsumedMessage(inflightMsg, topic, username) ); + m_interceptor.notifyMessageAcknowledged( new InterceptAcknowledgedMessage(inflightMsg, topic, username) ); } public void processDisconnect(Channel channel) { diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index f91e68dc..8a65fa4f 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -65,7 +65,7 @@ public class BrokerInterceptorTest { } @Override - public void onMessageConsumed( InterceptConsumedMessage msg ) { + public void onMessageAcknowledged( InterceptAcknowledgedMessage msg ) { n.set(90); } }