Add Chunk info consumer side (#355)

* Add chunk info on the consumer side
* Closes #331

---------
Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

RabbitMQ.Stream.Client/PublicAPI.Shipped.txt

@@ -395,7 +395,6 @@ RabbitMQ.Stream.Client.Message.Size.get -> int
 RabbitMQ.Stream.Client.Message.Write(System.Span<byte> span) -> int
 RabbitMQ.Stream.Client.MessageContext
 RabbitMQ.Stream.Client.MessageContext.MessageContext() -> void
-RabbitMQ.Stream.Client.MessageContext.MessageContext(ulong offset, System.TimeSpan timestamp) -> void
 RabbitMQ.Stream.Client.MessageContext.Offset.get -> ulong
 RabbitMQ.Stream.Client.MessageContext.Timestamp.get -> System.TimeSpan
 RabbitMQ.Stream.Client.MetaDataQuery

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -155,6 +155,9 @@ RabbitMQ.Stream.Client.ISuperStreamProducer.ReconnectPartition(RabbitMQ.Stream.C
 RabbitMQ.Stream.Client.KeyRoutingStrategy
 RabbitMQ.Stream.Client.KeyRoutingStrategy.KeyRoutingStrategy(System.Func<RabbitMQ.Stream.Client.Message, string> routingKeyExtractor, System.Func<string, string, System.Threading.Tasks.Task<RabbitMQ.Stream.Client.RouteQueryResponse>> routingKeyQFunc, string superStream) -> void
 RabbitMQ.Stream.Client.KeyRoutingStrategy.Route(RabbitMQ.Stream.Client.Message message, System.Collections.Generic.List<string> partitions) -> System.Threading.Tasks.Task<System.Collections.Generic.List<string>>
+RabbitMQ.Stream.Client.MessageContext.ChunkId.get -> ulong
+RabbitMQ.Stream.Client.MessageContext.ChunkMessagesCount.get -> uint
+RabbitMQ.Stream.Client.MessageContext.MessageContext(ulong offset, System.TimeSpan timestamp, uint chunkMessagesCount, ulong chunkId) -> void
 RabbitMQ.Stream.Client.OffsetTypeTimestamp.OffsetTypeTimestamp(System.DateTime dateTime) -> void
 RabbitMQ.Stream.Client.OffsetTypeTimestamp.OffsetTypeTimestamp(System.DateTimeOffset dateTimeOffset) -> void
 RabbitMQ.Stream.Client.ProducerFilter

RabbitMQ.Stream.Client/RawConsumer.cs

@@ -34,10 +34,22 @@ public struct MessageContext
         /// </summary>
         public TimeSpan Timestamp { get; }
 
-        public MessageContext(ulong offset, TimeSpan timestamp)
+        /// <summary>
+        /// The number of messages in the current chunk
+        /// </summary>
+        public uint ChunkMessagesCount { get; }
+
+        /// <summary>
+        /// It is the chunk id that can help to understand the ChuckMessagesCount
+        /// </summary>
+        public ulong ChunkId { get; }
+
+        public MessageContext(ulong offset, TimeSpan timestamp, uint chunkMessagesCount, ulong chunkId)
         {
             Offset = offset;
             Timestamp = timestamp;
+            ChunkMessagesCount = chunkMessagesCount;
+            ChunkId = chunkId;
         }
     }
 
@@ -306,7 +318,8 @@ async Task DispatchMessage(Message message, ulong i)
                                     {
                                         await _config.MessageHandler(this,
                                             new MessageContext(message.MessageOffset,
-                                                TimeSpan.FromMilliseconds(chunk.Timestamp)),
+                                                TimeSpan.FromMilliseconds(chunk.Timestamp),
+                                                chunk.NumRecords, chunk.ChunkId),
                                             message).ConfigureAwait(false);
                                     }
                                 }

Tests/RawConsumerSystemTests.cs

@@ -5,6 +5,7 @@
 using System;
 using System.Buffers;
 using System.Collections.Generic;
+using System.Linq;
 using System.Net;
 using System.Text;
 using System.Threading;
@@ -38,13 +39,15 @@ public async void CreateRawConsumer()
             var rawProducer = await system.CreateRawProducer(
                 new RawProducerConfig(stream) { Reference = "producer" });
             var identifierReceived = "";
+            uint messagesInTheChunk = 0;
             var consumer = await system.CreateRawConsumer(
                 new RawConsumerConfig(stream)
                 {
                     Reference = "consumer",
                     Identifier = "consumer_identifier_999",
                     MessageHandler = async (sourceConsumer, ctx, message) =>
                     {
+                        messagesInTheChunk = ctx.ChunkMessagesCount;
                         identifierReceived = sourceConsumer.Info.Identifier;
                         testPassed.SetResult(message.Data);
                         await Task.CompletedTask;
@@ -59,6 +62,7 @@ public async void CreateRawConsumer()
 
             Assert.Equal(msgData.Contents.ToArray(), testPassed.Task.Result.Contents.ToArray());
             Assert.Equal("consumer_identifier_999", identifierReceived);
+            Assert.Equal((uint)1, messagesInTheChunk);
             rawProducer.Dispose();
             consumer.Dispose();
             await system.DeleteStream(stream);
@@ -250,12 +254,15 @@ void AssertMessages(IReadOnlyList<Message> expected, IReadOnlyList<Message> actu
             await system.CreateStream(new StreamSpec(stream));
 
             var receivedMessages = new List<Message>();
+            var messagesInTheChunks = new Dictionary<ulong, uint>();
+
             var consumer = await system.CreateRawConsumer(
                 new RawConsumerConfig(stream)
                 {
                     Reference = "consumer",
                     MessageHandler = async (consumer, ctx, message) =>
                     {
+                        messagesInTheChunks[ctx.ChunkId] = ctx.ChunkMessagesCount;
                         receivedMessages.Add(message);
                         if (receivedMessages.Count == 10)
                         {
@@ -284,7 +291,8 @@ void AssertMessages(IReadOnlyList<Message> expected, IReadOnlyList<Message> actu
 
             AssertMessages(messagesGzip, testPassed.Task.Result.FindAll(s =>
                 Encoding.Default.GetString(s.Data.Contents.ToArray()).Contains("Gzip_")));
-
+            Assert.Equal(10, messagesInTheChunks.Sum(x => x.Value));
+            Assert.Equal((ulong)0, messagesInTheChunks.FirstOrDefault().Key);
             rawProducer.Dispose();
             consumer.Dispose();
             await system.DeleteStream(stream);

docs/ReliableClient/Program.cs

@@ -10,15 +10,15 @@
     ProducersPerConnection = 100, 
     ConsumersPerConnection = 100,
     Host = "node0",
-    Port = 5552,
-    LoadBalancer = true,
+    Port = 5562,
+    LoadBalancer = false,
     SuperStream = false,
     Streams = 3,
     Producers = 10,
-    MessagesPerProducer = 50_000_000,
+    MessagesPerProducer = 10_000_000,
     Consumers = 10,
-    Username = "test",
-    Password = "test"
+    // Username = "test",
+    // Password = "test"
 });
 
 await rClient.ConfigureAwait(false);