Added Message and Bytes read/write metrics

broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java

@@ -9,8 +9,7 @@ import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.handler.logging.LogLevel;
-import io.netty.handler.logging.LoggingHandler;
+
 import java.io.IOException;
 
 import io.netty.handler.timeout.IdleStateHandler;
@@ -19,6 +18,7 @@ import org.dna.mqtt.moquette.messaging.spi.IMessaging;
 import org.dna.mqtt.moquette.parser.netty.MQTTDecoder;
 import org.dna.mqtt.moquette.parser.netty.MQTTEncoder;
 import org.dna.mqtt.moquette.server.ServerAcceptor;
+import org.dna.mqtt.moquette.server.netty.metrics.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -32,7 +32,9 @@ public class NettyAcceptor implements ServerAcceptor {
     
     EventLoopGroup m_bossGroup;
     EventLoopGroup m_workerGroup;
-    
+    //BytesMetricsCollector m_metricsCollector = new BytesMetricsCollector();
+    MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
+
 
     public void initialize(IMessaging messaging) throws IOException {
         m_bossGroup = new NioEventLoopGroup();
@@ -40,19 +42,21 @@ public class NettyAcceptor implements ServerAcceptor {
         
         final NettyMQTTHandler handler = new NettyMQTTHandler();
         handler.setMessaging(messaging);
-        
-        ServerBootstrap b = new ServerBootstrap(); 
+
+        ServerBootstrap b = new ServerBootstrap();
             b.group(m_bossGroup, m_workerGroup)
              .channel(NioServerSocketChannel.class) 
              .childHandler(new ChannelInitializer<SocketChannel>() { 
                  @Override
                  public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline pipeline = ch.pipeline();
+                    //pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector));
                     pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
                     pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
                     //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
                     pipeline.addLast("decoder", new MQTTDecoder());
                     pipeline.addLast("encoder", new MQTTEncoder());
+                    pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
                     pipeline.addLast("handler", handler);
                  }
              })
@@ -78,6 +82,10 @@ public class NettyAcceptor implements ServerAcceptor {
         }
         m_workerGroup.shutdownGracefully();
         m_bossGroup.shutdownGracefully();
+
+        MessageMetrics metrics = m_metricsCollector.computeMetrics();
+        //LOG.info(String.format("Bytes read: %d, bytes wrote: %d", metrics.readBytes(), metrics.wroteBytes()));
+        LOG.info(String.format("Msg read: %d, msg wrote: %d", metrics.messagesRead(), metrics.messagesWrote()));
     }
     
 }

broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetrics.java

@@ -0,0 +1,22 @@
+package org.dna.mqtt.moquette.server.netty.metrics;
+
+public class BytesMetrics {
+    private long m_readBytes = 0;
+    private long m_wroteBytes = 0;
+
+    void incrementRead(long numBytes) {
+        m_readBytes += numBytes;
+    }
+
+    void incrementWrote(long numBytes) {
+        m_wroteBytes += numBytes;
+    }
+
+    public long readBytes() {
+        return m_readBytes;
+    }
+
+    public long wroteBytes() {
+        return m_wroteBytes;
+    }
+}

broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsCollector.java

@@ -0,0 +1,24 @@
+package org.dna.mqtt.moquette.server.netty.metrics;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Collects all the metrics from the various pipeline.
+ */
+public class BytesMetricsCollector {
+    private Queue<BytesMetrics> m_allMetrics = new ConcurrentLinkedQueue<BytesMetrics>();
+
+    void addMetrics(BytesMetrics metrics) {
+        m_allMetrics.add(metrics);
+    }
+
+    public BytesMetrics computeMetrics() {
+        BytesMetrics allMetrics = new BytesMetrics();
+        for (BytesMetrics m : m_allMetrics) {
+            allMetrics.incrementRead(m.readBytes());
+            allMetrics.incrementWrote(m.wroteBytes());
+        }
+        return allMetrics;
+    }
+}

broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/BytesMetricsHandler.java

@@ -0,0 +1,50 @@
+package org.dna.mqtt.moquette.server.netty.metrics;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+public class BytesMetricsHandler extends ChannelDuplexHandler {
+
+    private static final AttributeKey<BytesMetrics> ATTR_KEY_METRICS = new AttributeKey<BytesMetrics>("BytesMetrics");
+
+    private BytesMetricsCollector m_collector;
+
+    public BytesMetricsHandler(BytesMetricsCollector collector) {
+          m_collector = collector;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        Attribute<BytesMetrics> attr = ctx.attr(ATTR_KEY_METRICS);
+        attr.set(new BytesMetrics());
+
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
+        metrics.incrementRead(((ByteBuf)msg).readableBytes());
+        ctx.fireChannelRead(msg);
+    }
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
+        metrics.incrementWrote(((ByteBuf)msg).writableBytes());
+        ctx.write(msg, promise);
+    }
+
+
+    @Override
+    public void close(ChannelHandlerContext ctx,
+                      ChannelPromise promise) throws Exception {
+        BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
+        m_collector.addMetrics(metrics);
+        super.close(ctx, promise);
+    }
+}

broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetrics.java

@@ -0,0 +1,22 @@
+package org.dna.mqtt.moquette.server.netty.metrics;
+
+public class MessageMetrics {
+    private long m_messagesRead = 0;
+    private long m_messageWrote = 0;
+
+    void incrementRead(long numMessages) {
+        m_messagesRead += numMessages;
+    }
+
+    void incrementWrote(long numMessages) {
+        m_messageWrote += numMessages;
+    }
+
+    public long messagesRead() {
+        return m_messagesRead;
+    }
+
+    public long messagesWrote() {
+        return m_messageWrote;
+    }
+}

broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsCollector.java

@@ -0,0 +1,24 @@
+package org.dna.mqtt.moquette.server.netty.metrics;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Collects all the metrics from the various pipeline.
+ */
+public class MessageMetricsCollector {
+    private Queue<MessageMetrics> m_allMetrics = new ConcurrentLinkedQueue<MessageMetrics>();
+
+    void addMetrics(MessageMetrics metrics) {
+        m_allMetrics.add(metrics);
+    }
+
+    public MessageMetrics computeMetrics() {
+        MessageMetrics allMetrics = new MessageMetrics();
+        for (MessageMetrics m : m_allMetrics) {
+            allMetrics.incrementRead(m.messagesRead());
+            allMetrics.incrementWrote(m.messagesWrote());
+        }
+        return allMetrics;
+    }
+}

broker/src/main/java/org/dna/mqtt/moquette/server/netty/metrics/MessageMetricsHandler.java

@@ -0,0 +1,50 @@
+package org.dna.mqtt.moquette.server.netty.metrics;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+public class MessageMetricsHandler extends ChannelDuplexHandler {
+
+    private static final AttributeKey<MessageMetrics> ATTR_KEY_METRICS = new AttributeKey<MessageMetrics>("MessageMetrics");
+
+    private MessageMetricsCollector m_collector;
+
+    public MessageMetricsHandler(MessageMetricsCollector collector) {
+          m_collector = collector;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        Attribute<MessageMetrics> attr = ctx.attr(ATTR_KEY_METRICS);
+        attr.set(new MessageMetrics());
+
+        super.channelActive(ctx);
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
+        metrics.incrementRead(1);
+        ctx.fireChannelRead(msg);
+    }
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
+        metrics.incrementWrote(1);
+        ctx.write(msg, promise);
+    }
+
+
+    @Override
+    public void close(ChannelHandlerContext ctx,
+                      ChannelPromise promise) throws Exception {
+        MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
+        m_collector.addMetrics(metrics);
+        super.close(ctx, promise);
+    }
+}