Implement the super stream consumer Super Stream Consumer (#174)

Add IConsumer interface to be compliant to the Iproducer
The base for Standard Consumer and Super-Stream Consumer
Implement super-consumer low-level client
Implement the super stream for Reliable part


Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

Docker/Dockerfile

@@ -9,10 +9,12 @@ RUN wget https://packages.microsoft.com/config/ubuntu/21.04/packages-microsoft-p
 RUN dpkg -i packages-microsoft-prod.deb
 RUN rm packages-microsoft-prod.deb
 
-RUN apt-get update && \
-   apt-get install -y apt-transport-https && \
-   apt-get update && \
-   apt-get install -y git && \
-   apt-get install -y dotnet-sdk-6.0
+
+RUN printf 'Package: *net*\nPin: origin packages.microsoft.com\nPin-Priority: 1001' >> /etc/apt/preferences
+
+RUN apt-get update 
+RUN apt install dotnet6 -y
+RUN apt install git -y
+
 
 RUN apt-get install make -y

Makefile

@@ -22,4 +22,4 @@ run-test-in-docker:
 	docker stop dotnet-test || true
 	docker build -t stream-dotnet-test -f Docker/Dockerfile  . && \
 	docker run -d --rm --name dotnet-test -v $(shell pwd):/source --cpuset-cpus="1" stream-dotnet-test && \
-    docker exec -it dotnet-test /bin/sh -c "cd /source && make" || true  
+    docker exec -it dotnet-test /bin/sh -c "cd /source && make test" || true  

RabbitMQ.Stream.Client/Consumer.cs

@@ -34,15 +34,8 @@ public ConsumerEvents(Func<Deliver, Task> deliverHandler,
         public Func<bool, Task<IOffsetType>> ConsumerUpdateHandler { get; }
     }
 
-    public record ConsumerConfig : INamedEntity
+    public record ConsumerConfig : IConsumerConfig
     {
-        // StoredOffsetSpec configuration it is needed to keep the offset spec.
-        // since the offset can be decided from the ConsumerConfig.OffsetSpec.
-        // and from ConsumerConfig.ConsumerUpdateListener.
-        // needed also See also consumer:MaybeDispatch/1.
-        // It is not public because it is not needed for the user.
-        internal IOffsetType StoredOffsetSpec { get; set; }
-
         internal void Validate()
         {
             if (IsSingleActiveConsumer && (Reference == null || Reference.Trim() == string.Empty))
@@ -51,34 +44,17 @@ internal void Validate()
             }
         }
 
+        public IOffsetType OffsetSpec { get; set; } = new OffsetTypeNext();
+
         // stream name where the consumer will consume the messages.
         // stream must exist before the consumer is created.
         public string Stream { get; set; }
-        public string Reference { get; set; }
         public Func<Consumer, MessageContext, Message, Task> MessageHandler { get; set; }
-        public Func<string, Task> ConnectionClosedHandler { get; set; }
-
-        public IOffsetType OffsetSpec { get; set; } = new OffsetTypeNext();
-
-        // ClientProvidedName is used to identify TCP connection name.
-        public string ClientProvidedName { get; set; } = "dotnet-stream-consumer";
 
         public Action<MetaDataUpdate> MetadataHandler { get; set; } = _ => { };
-
-        // SingleActiveConsumer is used to indicate that there is only one consumer active for the stream.
-        // given a consumer reference. 
-        // Consumer Reference can't be null or Empty.
-
-        public bool IsSingleActiveConsumer { get; set; } = false;
-
-        // config.ConsumerUpdateListener is the callback for when the consumer is updated due
-        // to single active consumer. 
-        // return IOffsetType to indicate the offset to be used for the next consumption.
-        // if the ConsumerUpdateListener==null the OffsetSpec will be used.
-        public Func<string, string, bool, Task<IOffsetType>> ConsumerUpdateListener { get; set; } = null;
     }
 
-    public class Consumer : AbstractEntity, IDisposable
+    public class Consumer : AbstractEntity, IConsumer, IDisposable
     {
         private bool _disposed;
         private readonly ConsumerConfig config;
@@ -109,7 +85,7 @@ public async Task StoreOffset(ulong offset)
             await client.StoreOffset(config.Reference, config.Stream, offset);
         }
 
-        public static async Task<Consumer> Create(ClientParameters clientParameters,
+        public static async Task<IConsumer> Create(ClientParameters clientParameters,
             ConsumerConfig config,
             StreamInfo metaStreamInfo)
         {
@@ -159,11 +135,18 @@ private async Task Init()
                             continue;
                         }
 
-                        var message = Message.From(messageEntry.Data);
-                        await config.MessageHandler(this,
-                            new MessageContext(messageEntry.Offset,
-                                TimeSpan.FromMilliseconds(deliver.Chunk.Timestamp)),
-                            message);
+                        try
+                        {
+                            var message = Message.From(messageEntry.Data);
+                            await config.MessageHandler(this,
+                                new MessageContext(messageEntry.Offset,
+                                    TimeSpan.FromMilliseconds(deliver.Chunk.Timestamp)),
+                                message);
+                        }
+                        catch (Exception e)
+                        {
+                            LogEventSource.Log.LogError($"Error while processing message {messageEntry.Offset} {e}");
+                        }
                     }
 
                     // give one credit after each chunk

RabbitMQ.Stream.Client/IConsumer.cs

@@ -0,0 +1,43 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2007-2020 VMware, Inc.
+
+using System;
+using System.Threading.Tasks;
+
+namespace RabbitMQ.Stream.Client;
+
+public interface IConsumer
+{
+    public Task StoreOffset(ulong offset);
+    public Task<ResponseCode> Close();
+    public void Dispose();
+}
+
+public record IConsumerConfig : INamedEntity
+{
+    // StoredOffsetSpec configuration it is needed to keep the offset spec.
+    // since the offset can be decided from the ConsumerConfig.OffsetSpec.
+    // and from ConsumerConfig.ConsumerUpdateListener.
+    // needed also See also consumer:MaybeDispatch/1.
+    // It is not public because it is not needed for the user.
+    internal IOffsetType StoredOffsetSpec { get; set; }
+
+    // ClientProvidedName is used to identify TCP connection name.
+    public string ClientProvidedName { get; set; } = "dotnet-stream-consumer";
+
+    // SingleActiveConsumer is used to indicate that there is only one consumer active for the stream.
+    // given a consumer reference. 
+    // Consumer Reference can't be null or Empty.
+    public bool IsSingleActiveConsumer { get; set; } = false;
+
+    // config.ConsumerUpdateListener is the callback for when the consumer is updated due
+    // to single active consumer. 
+    // return IOffsetType to indicate the offset to be used for the next consumption.
+    // if the ConsumerUpdateListener==null the OffsetSpec will be used.
+    public Func<string, string, bool, Task<IOffsetType>> ConsumerUpdateListener { get; set; } = null;
+
+    public string Reference { get; set; }
+
+    public Func<string, Task> ConnectionClosedHandler { get; set; }
+}

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -278,22 +278,12 @@ RabbitMQ.Stream.Client.Consumer.Close() -> System.Threading.Tasks.Task<RabbitMQ.
 RabbitMQ.Stream.Client.Consumer.Dispose() -> void
 RabbitMQ.Stream.Client.Consumer.StoreOffset(ulong offset) -> System.Threading.Tasks.Task
 RabbitMQ.Stream.Client.ConsumerConfig
-RabbitMQ.Stream.Client.ConsumerConfig.ClientProvidedName.get -> string
-RabbitMQ.Stream.Client.ConsumerConfig.ClientProvidedName.set -> void
-RabbitMQ.Stream.Client.ConsumerConfig.ConnectionClosedHandler.get -> System.Func<string, System.Threading.Tasks.Task>
-RabbitMQ.Stream.Client.ConsumerConfig.ConnectionClosedHandler.set -> void
-RabbitMQ.Stream.Client.ConsumerConfig.ConsumerUpdateListener.get -> System.Func<string, string, bool, System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IOffsetType>>
-RabbitMQ.Stream.Client.ConsumerConfig.ConsumerUpdateListener.set -> void
-RabbitMQ.Stream.Client.ConsumerConfig.IsSingleActiveConsumer.get -> bool
-RabbitMQ.Stream.Client.ConsumerConfig.IsSingleActiveConsumer.set -> void
 RabbitMQ.Stream.Client.ConsumerConfig.MessageHandler.get -> System.Func<RabbitMQ.Stream.Client.Consumer, RabbitMQ.Stream.Client.MessageContext, RabbitMQ.Stream.Client.Message, System.Threading.Tasks.Task>
 RabbitMQ.Stream.Client.ConsumerConfig.MessageHandler.set -> void
 RabbitMQ.Stream.Client.ConsumerConfig.MetadataHandler.get -> System.Action<RabbitMQ.Stream.Client.MetaDataUpdate>
 RabbitMQ.Stream.Client.ConsumerConfig.MetadataHandler.set -> void
 RabbitMQ.Stream.Client.ConsumerConfig.OffsetSpec.get -> RabbitMQ.Stream.Client.IOffsetType
 RabbitMQ.Stream.Client.ConsumerConfig.OffsetSpec.set -> void
-RabbitMQ.Stream.Client.ConsumerConfig.Reference.get -> string
-RabbitMQ.Stream.Client.ConsumerConfig.Reference.set -> void
 RabbitMQ.Stream.Client.ConsumerConfig.Stream.get -> string
 RabbitMQ.Stream.Client.ConsumerConfig.Stream.set -> void
 RabbitMQ.Stream.Client.ConsumerUpdateQueryResponse
@@ -426,6 +416,21 @@ RabbitMQ.Stream.Client.ICompressionCodec.MessagesCount.get -> int
 RabbitMQ.Stream.Client.ICompressionCodec.UnCompress(System.Buffers.ReadOnlySequence<byte> source, uint dataLen, uint unCompressedDataSize) -> System.Buffers.ReadOnlySequence<byte>
 RabbitMQ.Stream.Client.ICompressionCodec.UnCompressedSize.get -> int
 RabbitMQ.Stream.Client.ICompressionCodec.Write(System.Span<byte> span) -> int
+RabbitMQ.Stream.Client.IConsumer
+RabbitMQ.Stream.Client.IConsumer.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+RabbitMQ.Stream.Client.IConsumer.Dispose() -> void
+RabbitMQ.Stream.Client.IConsumer.StoreOffset(ulong offset) -> System.Threading.Tasks.Task
+RabbitMQ.Stream.Client.IConsumerConfig
+RabbitMQ.Stream.Client.IConsumerConfig.ClientProvidedName.get -> string
+RabbitMQ.Stream.Client.IConsumerConfig.ClientProvidedName.set -> void
+RabbitMQ.Stream.Client.IConsumerConfig.ConnectionClosedHandler.get -> System.Func<string, System.Threading.Tasks.Task>
+RabbitMQ.Stream.Client.IConsumerConfig.ConnectionClosedHandler.set -> void
+RabbitMQ.Stream.Client.IConsumerConfig.ConsumerUpdateListener.get -> System.Func<string, string, bool, System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IOffsetType>>
+RabbitMQ.Stream.Client.IConsumerConfig.ConsumerUpdateListener.set -> void
+RabbitMQ.Stream.Client.IConsumerConfig.IsSingleActiveConsumer.get -> bool
+RabbitMQ.Stream.Client.IConsumerConfig.IsSingleActiveConsumer.set -> void
+RabbitMQ.Stream.Client.IConsumerConfig.Reference.get -> string
+RabbitMQ.Stream.Client.IConsumerConfig.Reference.set -> void
 RabbitMQ.Stream.Client.INamedEntity
 RabbitMQ.Stream.Client.INamedEntity.ClientProvidedName.get -> string
 RabbitMQ.Stream.Client.INamedEntity.ClientProvidedName.set -> void
@@ -703,6 +708,10 @@ RabbitMQ.Stream.Client.Reliable.ConfirmationStatus.PublisherDoesNotExist = 18 ->
 RabbitMQ.Stream.Client.Reliable.ConfirmationStatus.StreamNotAvailable = 6 -> RabbitMQ.Stream.Client.Reliable.ConfirmationStatus
 RabbitMQ.Stream.Client.Reliable.ConfirmationStatus.UndefinedError = 200 -> RabbitMQ.Stream.Client.Reliable.ConfirmationStatus
 RabbitMQ.Stream.Client.Reliable.ConfirmationStatus.WaitForConfirmation = 0 -> RabbitMQ.Stream.Client.Reliable.ConfirmationStatus
+RabbitMQ.Stream.Client.Reliable.ConsumerFactory
+RabbitMQ.Stream.Client.Reliable.ConsumerFactory.ConsumerFactory() -> void
+RabbitMQ.Stream.Client.Reliable.ConsumerFactory.CreateConsumer(bool boot) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IConsumer>
+RabbitMQ.Stream.Client.Reliable.ConsumerFactory._reliableConsumerConfig -> RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig
 RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
 RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenConnected(string connectionInfo) -> System.Threading.Tasks.ValueTask
 RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenDisconnected(string connectionInfo) -> System.Threading.Tasks.ValueTask<bool>
@@ -740,7 +749,9 @@ RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.ConsumerUpdateListener.ge
 RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.ConsumerUpdateListener.set -> void
 RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.IsSingleActiveConsumer.get -> bool
 RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.IsSingleActiveConsumer.set -> void
-RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.MessageHandler.get -> System.Func<RabbitMQ.Stream.Client.Consumer, RabbitMQ.Stream.Client.MessageContext, RabbitMQ.Stream.Client.Message, System.Threading.Tasks.Task>
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.IsSuperStream.get -> bool
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.IsSuperStream.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.MessageHandler.get -> System.Func<string, RabbitMQ.Stream.Client.Consumer, RabbitMQ.Stream.Client.MessageContext, RabbitMQ.Stream.Client.Message, System.Threading.Tasks.Task>
 RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.MessageHandler.set -> void
 RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.OffsetSpec.get -> RabbitMQ.Stream.Client.IOffsetType
 RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.OffsetSpec.set -> void
@@ -872,13 +883,15 @@ RabbitMQ.Stream.Client.StreamSpec.Name.init -> void
 RabbitMQ.Stream.Client.StreamSpec.StreamSpec(string Name) -> void
 RabbitMQ.Stream.Client.StreamSystem
 RabbitMQ.Stream.Client.StreamSystem.Close() -> System.Threading.Tasks.Task
-RabbitMQ.Stream.Client.StreamSystem.CreateConsumer(RabbitMQ.Stream.Client.ConsumerConfig consumerConfig) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Consumer>
+RabbitMQ.Stream.Client.StreamSystem.CreateConsumer(RabbitMQ.Stream.Client.ConsumerConfig consumerConfig) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IConsumer>
 RabbitMQ.Stream.Client.StreamSystem.CreateProducer(RabbitMQ.Stream.Client.ProducerConfig producerConfig) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IProducer>
 RabbitMQ.Stream.Client.StreamSystem.CreateStream(RabbitMQ.Stream.Client.StreamSpec spec) -> System.Threading.Tasks.Task
+RabbitMQ.Stream.Client.StreamSystem.CreateSuperStreamConsumer(RabbitMQ.Stream.Client.SuperStreamConsumerConfig superStreamConsumerConfig) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IConsumer>
 RabbitMQ.Stream.Client.StreamSystem.CreateSuperStreamProducer(RabbitMQ.Stream.Client.SuperStreamProducerConfig superStreamProducerConfig) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IProducer>
 RabbitMQ.Stream.Client.StreamSystem.DeleteStream(string stream) -> System.Threading.Tasks.Task
 RabbitMQ.Stream.Client.StreamSystem.IsClosed.get -> bool
 RabbitMQ.Stream.Client.StreamSystem.QueryOffset(string reference, string stream) -> System.Threading.Tasks.Task<ulong>
+RabbitMQ.Stream.Client.StreamSystem.QueryPartition(string superStream) -> System.Threading.Tasks.Task<string[]>
 RabbitMQ.Stream.Client.StreamSystem.QuerySequence(string reference, string stream) -> System.Threading.Tasks.Task<ulong>
 RabbitMQ.Stream.Client.StreamSystem.StreamExists(string stream) -> System.Threading.Tasks.Task<bool>
 RabbitMQ.Stream.Client.StreamSystemConfig
@@ -916,6 +929,17 @@ RabbitMQ.Stream.Client.SubscribeResponse.ResponseCode.get -> RabbitMQ.Stream.Cli
 RabbitMQ.Stream.Client.SubscribeResponse.SizeNeeded.get -> int
 RabbitMQ.Stream.Client.SubscribeResponse.SubscribeResponse() -> void
 RabbitMQ.Stream.Client.SubscribeResponse.Write(System.Span<byte> span) -> int
+RabbitMQ.Stream.Client.SuperStreamConsumer
+RabbitMQ.Stream.Client.SuperStreamConsumer.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+RabbitMQ.Stream.Client.SuperStreamConsumer.Dispose() -> void
+RabbitMQ.Stream.Client.SuperStreamConsumer.StoreOffset(ulong offset) -> System.Threading.Tasks.Task
+RabbitMQ.Stream.Client.SuperStreamConsumerConfig
+RabbitMQ.Stream.Client.SuperStreamConsumerConfig.MessageHandler.get -> System.Func<string, RabbitMQ.Stream.Client.Consumer, RabbitMQ.Stream.Client.MessageContext, RabbitMQ.Stream.Client.Message, System.Threading.Tasks.Task>
+RabbitMQ.Stream.Client.SuperStreamConsumerConfig.MessageHandler.set -> void
+RabbitMQ.Stream.Client.SuperStreamConsumerConfig.OffsetSpec.get -> System.Collections.Concurrent.ConcurrentDictionary<string, RabbitMQ.Stream.Client.IOffsetType>
+RabbitMQ.Stream.Client.SuperStreamConsumerConfig.OffsetSpec.set -> void
+RabbitMQ.Stream.Client.SuperStreamConsumerConfig.SuperStream.get -> string
+RabbitMQ.Stream.Client.SuperStreamConsumerConfig.SuperStream.set -> void
 RabbitMQ.Stream.Client.SuperStreamProducer
 RabbitMQ.Stream.Client.SuperStreamProducer.BatchSend(System.Collections.Generic.List<(ulong, RabbitMQ.Stream.Client.Message)> messages) -> System.Threading.Tasks.ValueTask
 RabbitMQ.Stream.Client.SuperStreamProducer.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
@@ -978,7 +1002,7 @@ static RabbitMQ.Stream.Client.Client.Create(RabbitMQ.Stream.Client.ClientParamet
 static RabbitMQ.Stream.Client.CompressionHelper.Compress(System.Collections.Generic.List<RabbitMQ.Stream.Client.Message> messages, RabbitMQ.Stream.Client.CompressionType compressionType) -> RabbitMQ.Stream.Client.ICompressionCodec
 static RabbitMQ.Stream.Client.CompressionHelper.UnCompress(RabbitMQ.Stream.Client.CompressionType compressionType, System.Buffers.ReadOnlySequence<byte> source, uint dataLen, uint unCompressedDataSize) -> System.Buffers.ReadOnlySequence<byte>
 static RabbitMQ.Stream.Client.Connection.Create(System.Net.EndPoint endpoint, System.Func<System.Memory<byte>, System.Threading.Tasks.Task> commandCallback, System.Func<string, System.Threading.Tasks.Task> closedCallBack, RabbitMQ.Stream.Client.SslOption sslOption) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Connection>
-static RabbitMQ.Stream.Client.Consumer.Create(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.ConsumerConfig config, RabbitMQ.Stream.Client.StreamInfo metaStreamInfo) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Consumer>
+static RabbitMQ.Stream.Client.Consumer.Create(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.ConsumerConfig config, RabbitMQ.Stream.Client.StreamInfo metaStreamInfo) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IConsumer>
 static RabbitMQ.Stream.Client.LeaderLocator.ClientLocal.get -> RabbitMQ.Stream.Client.LeaderLocator
 static RabbitMQ.Stream.Client.LeaderLocator.LeastLeaders.get -> RabbitMQ.Stream.Client.LeaderLocator
 static RabbitMQ.Stream.Client.LeaderLocator.Random.get -> RabbitMQ.Stream.Client.LeaderLocator
@@ -996,5 +1020,6 @@ static RabbitMQ.Stream.Client.StreamCompressionCodecs.RegisterCodec<T>(RabbitMQ.
 static RabbitMQ.Stream.Client.StreamCompressionCodecs.UnRegisterCodec(RabbitMQ.Stream.Client.CompressionType compressionType) -> void
 static RabbitMQ.Stream.Client.StreamSystem.Create(RabbitMQ.Stream.Client.StreamSystemConfig config) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.StreamSystem>
 static RabbitMQ.Stream.Client.SubEntryPublish.Version.get -> byte
+static RabbitMQ.Stream.Client.SuperStreamConsumer.Create(RabbitMQ.Stream.Client.SuperStreamConsumerConfig superStreamConsumerConfig, System.Collections.Generic.IDictionary<string, RabbitMQ.Stream.Client.StreamInfo> streamInfos, RabbitMQ.Stream.Client.ClientParameters clientParameters) -> RabbitMQ.Stream.Client.IConsumer
 static RabbitMQ.Stream.Client.SuperStreamProducer.Create(RabbitMQ.Stream.Client.SuperStreamProducerConfig superStreamProducerConfig, System.Collections.Generic.IDictionary<string, RabbitMQ.Stream.Client.StreamInfo> streamInfos, RabbitMQ.Stream.Client.ClientParameters clientParameters) -> RabbitMQ.Stream.Client.IProducer
 static RabbitMQ.Stream.Client.Version.VersionString.get -> string