From e7f26e402b6972bbb43c69acdabaa7253810b0fe Mon Sep 17 00:00:00 2001 From: Luis Barrios Date: Thu, 12 Jan 2017 17:25:46 +0100 Subject: [PATCH] Include message and bytes metrics in the status of the active MQTT connections. --- .../connections/MqttConnectionMetrics.java | 38 +++++++++++++++++++ .../io/moquette/connections/MqttSession.java | 33 ++++++++++------ .../moquette/server/ConnectionDescriptor.java | 16 +++++++- .../server/ConnectionDescriptorStore.java | 14 ++++++- .../netty/metrics/BytesMetricsHandler.java | 5 +++ .../netty/metrics/MessageMetricsHandler.java | 5 +++ 6 files changed, 96 insertions(+), 15 deletions(-) create mode 100644 broker/src/main/java/io/moquette/connections/MqttConnectionMetrics.java diff --git a/broker/src/main/java/io/moquette/connections/MqttConnectionMetrics.java b/broker/src/main/java/io/moquette/connections/MqttConnectionMetrics.java new file mode 100644 index 00000000..5ec7c267 --- /dev/null +++ b/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; + } + +} diff --git a/broker/src/main/java/io/moquette/connections/MqttSession.java b/broker/src/main/java/io/moquette/connections/MqttSession.java index 5d5082e9..6045ebcf 100644 --- a/broker/src/main/java/io/moquette/connections/MqttSession.java +++ b/broker/src/main/java/io/moquette/connections/MqttSession.java @@ -18,6 +18,7 @@ public class MqttSession { private int pendingPublishMessagesNo; private int secondPhaseAckPendingMessages; private Collection 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 getActiveSubscriptions() { - return activeSubscriptions; - } + public Collection getActiveSubscriptions() { + return activeSubscriptions; + } - public void setActiveSubscriptions(Collection activeSubscriptions) { - this.activeSubscriptions = activeSubscriptions; - } + public void setActiveSubscriptions(Collection activeSubscriptions) { + this.activeSubscriptions = activeSubscriptions; + } + + public MqttConnectionMetrics getConnectionMetrics() { + return connectionMetrics; + } + + public void setConnectionMetrics(MqttConnectionMetrics connectionMetrics) { + this.connectionMetrics = connectionMetrics; + } } diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 6e3c5c62..ddad36e0 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/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; diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java index 732512ef..7b0ffc39 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java +++ b/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()); diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java index 938b3d20..e948b3b8 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/BytesMetricsHandler.java +++ b/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(); + } } diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java index b11f0ee4..7786a5af 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MessageMetricsHandler.java +++ b/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(); + } }