Fixed issue #41 memory leak on metrics collectors

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

@@ -15,25 +15,28 @@
  */
 package org.eclipse.moquette.server.netty.metrics;
 
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * 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);
-    }
+    private AtomicLong readBytes = new AtomicLong();
+    private AtomicLong wroteBytes = new AtomicLong();
 
     public BytesMetrics computeMetrics() {
         BytesMetrics allMetrics = new BytesMetrics();
-        for (BytesMetrics m : m_allMetrics) {
-            allMetrics.incrementRead(m.readBytes());
-            allMetrics.incrementWrote(m.wroteBytes());
-        }
+        allMetrics.incrementRead(readBytes.get());
+        allMetrics.incrementWrote(wroteBytes.get());
         return allMetrics;
     }
+
+    public void sumReadBytes(long count) {
+        readBytes.getAndAdd(count);
+    }
+
+    public void sumWroteBytes(long count) {
+        wroteBytes.getAndAdd(count);
+    }
 }

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

@@ -59,7 +59,8 @@ public class BytesMetricsHandler extends ChannelDuplexHandler {
     public void close(ChannelHandlerContext ctx,
                       ChannelPromise promise) throws Exception {
         BytesMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
-        m_collector.addMetrics(metrics);
+        m_collector.sumReadBytes(metrics.readBytes());
+        m_collector.sumWroteBytes(metrics.wroteBytes());
         super.close(ctx, promise);
     }
 }

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

@@ -17,23 +17,27 @@ package org.eclipse.moquette.server.netty.metrics;
 
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * 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);
-    }
+    private AtomicLong readMsgs = new AtomicLong();
+    private AtomicLong wroteMsgs = new AtomicLong();
 
     public MessageMetrics computeMetrics() {
         MessageMetrics allMetrics = new MessageMetrics();
-        for (MessageMetrics m : m_allMetrics) {
-            allMetrics.incrementRead(m.messagesRead());
-            allMetrics.incrementWrote(m.messagesWrote());
-        }
+        allMetrics.incrementRead(readMsgs.get());
+        allMetrics.incrementWrote(wroteMsgs.get());
         return allMetrics;
     }
+
+    public void sumReadMessages(long count) {
+        readMsgs.getAndAdd(count);
+    }
+
+    public void sumWroteMessages(long count) {
+        wroteMsgs.getAndAdd(count);
+    }
 }

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

@@ -58,7 +58,8 @@ public class MessageMetricsHandler extends ChannelDuplexHandler {
     public void close(ChannelHandlerContext ctx,
                       ChannelPromise promise) throws Exception {
         MessageMetrics metrics = ctx.attr(ATTR_KEY_METRICS).get();
-        m_collector.addMetrics(metrics);
+        m_collector.sumReadMessages(metrics.messagesRead());
+        m_collector.sumWroteMessages(metrics.messagesWrote());
         super.close(ctx, promise);
     }
 }