anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Based on feedback, renamed messageConsumed -> messageAcknowledged.

Browse Source
This commit is contained in:
Dee Eldredge
2016-03-23 23:11:09 +01:00
committed by andsel
parent 1b77e43b61
commit 1e7f6d98e7
7 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java
+1 -1
View File
@@ -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) {}
}
broker/src/main/java/io/moquette/interception/InterceptHandler.java
+2 -2
View File
@@ -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);
}
broker/src/main/java/io/moquette/interception/Interceptor.java
+2 -2
View File
@@ -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);
}
broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java → broker/src/main/java/io/moquette/interception/messages/InterceptAcknowledgedMessage.java
+2 -2
View File
@@ -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;
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+2 -2
View File
@@ -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);
}
});
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+3 -3
View File
@@ -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) {