From 1b77e43b6156bdafd0769b261c45c920dd2b4a10 Mon Sep 17 00:00:00 2001
From: Dee Eldredge <Dee.Eldredge@VerizonWireless.com>
Date: Thu, 3 Mar 2016 18:43:40 -0800
Subject: [PATCH] Added topic to consumed message.

---
 .../messages/InterceptConsumedMessage.java             | 10 ++++++++--
 .../java/io/moquette/spi/impl/ProtocolProcessor.java   |  7 +++++--
 2 files changed, 13 insertions(+), 4 deletions(-)

diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java
index 9e73c832..0a01c9bd 100644
--- a/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java
+++ b/broker/src/main/java/io/moquette/interception/messages/InterceptConsumedMessage.java
@@ -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;
+	}
 }
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 a614aa3c..af14bc3a 100644
--- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
@@ -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) {