Collect consumed metric message by message

Instead of collecting after the whole chunk is dispatched. This makes
the metric more accurate when message processing takes some time.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/ServerFrameHandler.java

@@ -451,7 +451,6 @@ static int handleDeliver(
       }
 
       metricsCollector.chunk(numEntries);
-      long messagesRead = 0;
       MutableBoolean messageIgnored = new MutableBoolean(false);
 
       while (numRecords != 0) {
@@ -482,7 +481,7 @@ static int handleDeliver(
                 subscriptionId, offset, chunkTimestamp, committedOffset, chunkContext);
             messageIgnored.set(false);
           } else {
-            messagesRead++;
+            metricsCollector.consume(1);
           }
           numRecords--;
           offset++; // works even for unsigned long
@@ -551,7 +550,7 @@ static int handleDeliver(
                   subscriptionId, offset, chunkTimestamp, committedOffset, chunkContext);
               messageIgnored.set(false);
             } else {
-              messagesRead++;
+              metricsCollector.consume(1);
             }
             numRecordsInBatch--;
             offset++; // works even for unsigned long
@@ -564,7 +563,6 @@ static int handleDeliver(
           }
         }
       }
-      metricsCollector.consume(messagesRead);
       return read;
     }