Include message and bytes metrics in the status of the active MQTT connections.

broker/src/main/java/io/moquette/connections/MqttConnectionMetrics.java

@@ -0,0 +1,38 @@
+package io.moquette.connections;
+
+/**
+ * A class that represents the metrics of a given MQTT connection.
+ * 
+ * @author lbarrios
+ *
+ */
+public class MqttConnectionMetrics {
+    private long readKb;
+    private long writtenKb;
+    private long readMessages;
+    private long writtenMessages;
+
+    public MqttConnectionMetrics(long readBytes, long writtenBytes, long readMessages, long writtenMessages) {
+        this.readKb = readBytes / 1024;
+        this.writtenKb = writtenBytes / 1024;
+        this.readMessages = readMessages;
+        this.writtenMessages = writtenMessages;
+    }
+
+    public long getReadKb() {
+        return readKb;
+    }
+
+    public long getWrittenKb() {
+        return writtenKb;
+    }
+
+    public long getReadMessages() {
+        return readMessages;
+    }
+
+    public long getWrittenMessages() {
+        return writtenMessages;
+    }
+
+}

broker/src/main/java/io/moquette/connections/MqttSession.java

@@ -18,6 +18,7 @@ public class MqttSession {
 	private int pendingPublishMessagesNo;
 	private int secondPhaseAckPendingMessages;
 	private Collection<MqttSubscription> activeSubscriptions;
+	private MqttConnectionMetrics connectionMetrics;
 
 	public boolean isConnectionEstablished() {
 		return connectionEstablished;
@@ -51,20 +52,28 @@ public class MqttSession {
 		this.inflightMessages = inflightMessages;
 	}
 
-	public int getSecondPhaseAckPendingMessages() {
-		return secondPhaseAckPendingMessages;
-	}
+    public int getSecondPhaseAckPendingMessages() {
+        return secondPhaseAckPendingMessages;
+    }
 
-	public void setSecondPhaseAckPendingMessages(int secondPhaseAckPendingMessages) {
-		this.secondPhaseAckPendingMessages = secondPhaseAckPendingMessages;
-	}
+    public void setSecondPhaseAckPendingMessages(int secondPhaseAckPendingMessages) {
+        this.secondPhaseAckPendingMessages = secondPhaseAckPendingMessages;
+    }
 
-	public Collection<MqttSubscription> getActiveSubscriptions() {
-		return activeSubscriptions;
-	}
+    public Collection<MqttSubscription> getActiveSubscriptions() {
+        return activeSubscriptions;
+    }
 
-	public void setActiveSubscriptions(Collection<MqttSubscription> activeSubscriptions) {
-		this.activeSubscriptions = activeSubscriptions;
-	}
+    public void setActiveSubscriptions(Collection<MqttSubscription> activeSubscriptions) {
+        this.activeSubscriptions = activeSubscriptions;
+    }
+
+    public MqttConnectionMetrics getConnectionMetrics() {
+        return connectionMetrics;
+    }
+
+    public void setConnectionMetrics(MqttConnectionMetrics connectionMetrics) {
+        this.connectionMetrics = connectionMetrics;
+    }
 
 }

broker/src/main/java/io/moquette/server/ConnectionDescriptor.java

@@ -15,16 +15,20 @@
  */
 package io.moquette.server;
 
-import io.netty.channel.Channel;
 import java.util.NoSuchElementException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.atomic.AtomicReference;
 import io.moquette.server.netty.AutoFlushHandler;
 import io.moquette.server.netty.NettyUtils;
+import io.moquette.server.netty.metrics.BytesMetrics;
+import io.moquette.server.netty.metrics.BytesMetricsHandler;
+import io.moquette.server.netty.metrics.MessageMetrics;
+import io.moquette.server.netty.metrics.MessageMetricsHandler;
+import io.netty.channel.Channel;
 
 
 /**
@@ -129,6 +133,14 @@ public class ConnectionDescriptor {
 
     }
 
+    public BytesMetrics getBytesMetrics() {
+        return BytesMetricsHandler.getBytesMetrics(channel);
+    }
+
+    public MessageMetrics getMessageMetrics() {
+        return MessageMetricsHandler.getMessageMetrics(channel);
+    }
+
     @Override
     public int hashCode() {
         int result = clientID != null ? clientID.hashCode() : 0;

broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java

@@ -9,9 +9,12 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.moquette.connections.IConnectionsManager;
+import io.moquette.connections.MqttConnectionMetrics;
 import io.moquette.connections.MqttSession;
 import io.moquette.connections.MqttSubscription;
 import io.moquette.parser.proto.messages.AbstractMessage;
+import io.moquette.server.netty.metrics.BytesMetrics;
+import io.moquette.server.netty.metrics.MessageMetrics;
 import io.moquette.spi.ClientSession;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.subscriptions.Subscription;
@@ -150,7 +153,16 @@ public class ConnectionDescriptorStore implements IConnectionsManager {
         }
         result.setActiveSubscriptions(mqttSubscriptions);
         result.setCleanSession(session.isCleanSession());
-        result.setConnectionEstablished(this.isConnected(session.clientID));
+        ConnectionDescriptor descriptor = this.getConnection(session.clientID);
+        if (descriptor != null) {
+            result.setConnectionEstablished(true);
+            BytesMetrics bytesMetrics = descriptor.getBytesMetrics();
+            MessageMetrics messageMetrics = descriptor.getMessageMetrics();
+            result.setConnectionMetrics(new MqttConnectionMetrics(bytesMetrics.readBytes(), bytesMetrics.wroteBytes(),
+                    messageMetrics.messagesRead(), messageMetrics.messagesWrote()));
+        } else {
+            result.setConnectionEstablished(false);
+        }
         result.setPendingPublishMessagesNo(session.getPendingPublishMessagesNo());
         result.setSecondPhaseAckPendingMessages(session.getSecondPhaseAckPendingMessages());
         result.setInflightMessages(session.getInflightMessagesNo());

broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java

@@ -16,6 +16,7 @@
 package io.moquette.server.netty.metrics;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
@@ -63,4 +64,8 @@ public class BytesMetricsHandler extends ChannelDuplexHandler {
         m_collector.sumWroteBytes(metrics.wroteBytes());
         super.close(ctx, promise);
     }
+
+    public static BytesMetrics getBytesMetrics(Channel channel) {
+        return channel.attr(ATTR_KEY_METRICS).get();
+    }
 }

broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java

@@ -15,6 +15,7 @@
  */
 package io.moquette.server.netty.metrics;
 
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
@@ -62,4 +63,8 @@ public class MessageMetricsHandler extends ChannelDuplexHandler {
         m_collector.sumWroteMessages(metrics.messagesWrote());
         super.close(ctx, promise);
     }
+
+    public static MessageMetrics getMessageMetrics(Channel channel) {
+        return channel.attr(ATTR_KEY_METRICS).get();
+    }
 }