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

Added topic to consumed message.

Browse Source
This commit is contained in:
Dee Eldredge
2016-03-23 23:10:45 +01:00
committed by andsel
parent 7dab8a367f
commit 1b77e43b61
2 changed files with 13 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java
+8 -2
View File
@@ -4,11 +4,13 @@ import static io.moquette.spi.IMessagesStore.StoredMessage;
public class InterceptConsumedMessage {
final private StoredMessage msg;
private final String username;
private final String username;
private final String topic;
public InterceptConsumedMessage( final StoredMessage msg, String username) {
public InterceptConsumedMessage( final StoredMessage msg, final String topic, final String username) {
this.msg = msg;
this.username = username;
this.topic = topic;
}
public StoredMessage getMsg() {
@@ -18,4 +20,8 @@ public class InterceptConsumedMessage {
public String getUsername() {
return username;
}
public String getTopic() {
return topic;
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+5 -2
View File
@@ -287,8 +287,10 @@ public class ProtocolProcessor {
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
verifyToActivate(clientID, targetSession);
targetSession.inFlightAcknowledged(messageID);
String topic = inflightMsg.getTopic();
m_interceptor.notifyMessageConsumed(new InterceptConsumedMessage(inflightMsg, username));
m_interceptor.notifyMessageConsumed(new InterceptConsumedMessage(inflightMsg, topic, username));
}
private void verifyToActivate(String clientID, ClientSession targetSession) {
@@ -587,7 +589,8 @@ public class ProtocolProcessor {
verifyToActivate(clientID, targetSession);
targetSession.secondPhaseAcknowledged(messageID);
String username = NettyUtils.userName(channel);
m_interceptor.notifyMessageConsumed( new InterceptConsumedMessage(inflightMsg, username) );
String topic = inflightMsg.getTopic();
m_interceptor.notifyMessageConsumed( new InterceptConsumedMessage(inflightMsg, topic, username) );
}
public void processDisconnect(Channel channel) {