diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java b/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java index 42626b6d..0a846f0a 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/metrics/DropWizardMetricsHandler.java @@ -15,10 +15,12 @@ */ package io.moquette.server.netty.metrics; +import com.codahale.metrics.Counter; import com.codahale.metrics.Meter; import com.codahale.metrics.MetricRegistry; import com.librato.metrics.reporter.Librato; import io.moquette.server.config.IConfig; +import io.moquette.server.netty.NettyUtils; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.mqtt.MqttMessage; @@ -26,9 +28,7 @@ import io.netty.handler.codec.mqtt.MqttMessageType; import java.util.concurrent.TimeUnit; -import static io.moquette.BrokerConstants.METRICS_LIBRATO_EMAIL_PROPERTY_NAME; -import static io.moquette.BrokerConstants.METRICS_LIBRATO_TOKEN_PROPERTY_NAME; -import static io.moquette.BrokerConstants.METRICS_LIBRATO_SOURCE_PROPERTY_NAME; +import static io.moquette.BrokerConstants.*; import static io.netty.channel.ChannelHandler.Sharable; /** @@ -36,12 +36,16 @@ import static io.netty.channel.ChannelHandler.Sharable; */ @Sharable public final class DropWizardMetricsHandler extends ChannelInboundHandlerAdapter { - private MetricRegistry metrics = new MetricRegistry(); - private Meter publishesMetrics = metrics.meter("publish.requests"); + private MetricRegistry metrics; + private Meter publishesMetrics; + private Meter subscribeMetrics; + private Counter connectedClientsMetrics; public void init(IConfig props) { this.metrics = new MetricRegistry(); this.publishesMetrics = metrics.meter("publish.requests"); + this.subscribeMetrics = metrics.meter("subscribe.requests"); + this.connectedClientsMetrics = metrics.counter("connect.num_clients"); // ConsoleReporter reporter = ConsoleReporter.forRegistry(metrics) // .convertRatesTo(TimeUnit.SECONDS) // .convertDurationsTo(TimeUnit.MILLISECONDS) @@ -62,7 +66,16 @@ public final class DropWizardMetricsHandler extends ChannelInboundHandlerAdapter MqttMessageType messageType = msg.fixedHeader().messageType(); switch (messageType) { case PUBLISH: - publishesMetrics.mark(); + this.publishesMetrics.mark(); + break; + case SUBSCRIBE: + this.subscribeMetrics.mark(); + break; + case CONNECT: + this.connectedClientsMetrics.inc(); + break; + case DISCONNECT: + this.connectedClientsMetrics.dec(); break; default: break; @@ -70,4 +83,13 @@ public final class DropWizardMetricsHandler extends ChannelInboundHandlerAdapter ctx.fireChannelRead(message); } + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + String clientID = NettyUtils.clientID(ctx.channel()); + if (clientID != null && !clientID.isEmpty()) { + this.connectedClientsMetrics.dec(); + } + ctx.fireChannelInactive(); + } + }