reliable consumer (#115)

* reliable consumer work in progress

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

* reliable consumer

Add tests
Refactor adding the base class for Rproducer and Rconsumer

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

* reliable consumer

Add tests
Refactor adding the base class for Rproducer and Rconsumer

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

* Add R Producer Documentation [skip ci]

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

* Add new test

Fix a small bug during the reconnection

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

* formatting

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

* Increse timeout

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

* formatting

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

* update unshipped

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

* avoid timeout test

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

* avoid timeout test

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

* Add logs

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

* Wait on conditions rather than a set number of seconds

* fixup

* Fixup test

* Try to fix test on GHA

* Add logs

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

* Fix handler == null

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

Fix handler == null

* Escape HTTP connection name

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

* new Uri() should do the right thing

* fixup

* fixup

* Check HTTP return code consistently

* Add code to see what is going on with those connection names and encoding

* Use Uri.EscapeDataString

* Change the IReconnectStrategy

method. Use a function istead of out var

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

Co-authored-by: Luke Bakken <[email protected]>

Author: Gsantomaggio

Docker/Dockerfile

@@ -1,12 +1,18 @@
 FROM  pivotalrabbitmq/rabbitmq-stream
 
+
+ENV TZ=Europe/Minsk
+RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
+
+
 RUN wget https://packages.microsoft.com/config/ubuntu/21.04/packages-microsoft-prod.deb -O packages-microsoft-prod.deb
 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 apt-get install make -y

Examples/Reliable/ReliableConsumer.cs

@@ -0,0 +1,34 @@
+using System.Buffers;
+using System.Text;
+using RabbitMQ.Stream.Client;
+using RabbitMQ.Stream.Client.Reliable;
+
+namespace example;
+
+public class ReliableConsumerExample
+{
+    public async Task StartDefaultConfigurations()
+    {
+        Console.WriteLine("Reliable .NET Cosumer");
+        var config = new StreamSystemConfig();
+        const string stream = "my-reliable-stream-consumer";
+        var system = await StreamSystem.Create(config);
+        await system.CreateStream(new StreamSpec(stream));
+        var rConsumer = await ReliableConsumer.CreateReliableConsumer(new ReliableConsumerConfig()
+        {
+            Stream = stream,
+            StreamSystem = system,
+            OffsetSpec = new OffsetTypeFirst(),
+            ClientProvidedName = "My-Reliable-Consumer",
+            MessageHandler = async (_, _, message) =>
+            {
+                Console.WriteLine(
+                    $"message: {Encoding.Default.GetString(message.Data.Contents.ToArray())}");
+                await Task.CompletedTask;
+            }
+        });
+    }
+    
+    // when finished:
+    //rConsumer.Close();
+}

Examples/Reliable/ReliableProducer.cs

@@ -6,7 +6,7 @@
 
 namespace example;
 
-public class ReliableProducer
+public class ReliableProducerExample
 {
     public async Task StartDefaultConfigurations()
     {

Makefile

@@ -21,5 +21,5 @@ rabbitmq-server:
 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="0-1" stream-dotnet-test && \
+	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  

README.md

@@ -31,6 +31,7 @@
     - [Handle Metadata Update](#handle-metadata-update)
     - [Reliable](#reliable)
       - [Reliable Producer](#reliable-producer)
+      - [Reliable Consumer](#reliable-consumer)
 - [Build from source](#build-from-source)
 - [Project Status](#project-status)
 - [Release Process](#release-process)
@@ -412,8 +413,10 @@ You can use `MetadataHandler` to handle it:
 ```
 ### Reliable 
  - Reliable Producer
- - Reliable Consumer (not ready yet)
+ - Reliable Consumer </p>
+See the directory [Examples/Reliable](./Examples/Reliable)
 
+ 
 ### Reliable Producer
 Reliable Producer is a smart layer built up of the standard `Producer`. </b>   
 The idea is to leave the user decides what to use, the standard or reliable producer. </b>
@@ -423,7 +426,7 @@ The main features are:
 - Auto-Reconnect in case of disconnection
 - Trace sent and received messages
 - Invalidate messages
-- Handle the metadata Update
+- [Handle the metadata Update](#reliable-handle-metadata-update)
 
 #### Provide publishingID automatically
 Reliable Producer retrieves the last publishingID given the producer name. </b>
@@ -433,6 +436,7 @@ Zero(0) is the default value in case there is no a publishingID.
 Reliable Producer restores the TCP connection in case the Producer is disconnected for some reason.
 During the reconnection it continues to store the messages in a local-list.
 The user will receive back the confirmed or un-confirmed messages.
+See [Reconnection Strategy](#reconnection-strategy)
 
 #### Trace sent and received messages
 Reliable Producer keeps in memory each sent message and remove from the memory when the message is confirmed or goes in timout.
@@ -457,22 +461,35 @@ ConfirmationHandler = confirmation =>
 If the client doesn't receive a confirmation within 2 seconds Reliable Producer removes the message from the internal messages cache.
 The user will receive `ConfirmationStatus.TimeoutError` in the `ConfirmationHandler`.
 
-#### Handle the metadata Update
-If the streams changes the topology (ex:Stream deleted or add/remove follower), the client receives an `MetadataUpdate` event.
-Reliable Producer detects the event and tries to reconnect the producer if the stream still exist else closes the producer.
 #### Send API
 Reliable Producer implements two `send(..)`
 - `Send(Message message)` // standard
 - `Send(List<Message> messages, CompressionType compressionType)` //sub-batching with compression
 
-### Reconnection Strategy 
-By default Reliable Producer uses an `BackOffReconnectStrategy` to reconnect the client.
+
+### Reliable Consumer
+Reliable Consumer is a smart layer built up of the standard `Consumer`. </b>   
+The idea is to leave the user decides what to use, the standard or reliable Consumer. </b>
+
+The main features are:
+- Auto-Reconnect in case of disconnection
+- Auto restart consuming from the last offset
+- [Handle the metadata Update](#reliable-handle-metadata-update)
+
+#### Auto-Reconnect
+Reliable Consumer restores the TCP connection in case the Producer is disconnected for some reason.
+Reliable Consumer will restart consuming from the last offset stored.
+See [Reconnection Strategy](#reconnection-strategy)
+
+### Reconnection Strategy
+By default Reliable Producer/Consumer uses an `BackOffReconnectStrategy` to reconnect the client.
 You can customize the behaviour implementing the `IReconnectStrategy` interface:
 ```csharp
-void WhenDisconnected(out bool reconnect);
-void WhenConnected();
+bool WhenDisconnected(string connectionInfo);
+void WhenConnected(string connectionInfo);
 ```
-with `reconnect` you can decide when reconnect the producer.
+If `WhenDisconnected` return is `true` Producer/Consumer will be reconnected else closed.
+`connectionInfo` add information about the connection.
 
 You can use it:
 ```csharp
@@ -482,9 +499,10 @@ var p = await ReliableProducer.CreateReliableProducer(new ReliableProducerConfig
 ReconnectStrategy = MyReconnectStrategy
 ```
 
-#### Examples
-See the directory [Examples/Reliable](./Examples/Reliable)
 
+### Reliable handle metadata update
+If the streams changes the topology (ex:Stream deleted or add/remove follower), the client receives an `MetadataUpdate` event.
+Reliable Producer detects the event and tries to reconnect the producer if the stream still exist else closes the producer/consumer.
 
 ## Build from source
 

RabbitMQ.Stream.Client/AMQP/AmqpWireFormattingWrite.cs

@@ -29,13 +29,11 @@ public static int WriteAny(Span<byte> seq, object value)
                 bool bo => WriteBool(seq, bo),
                 byte[] bArr => bArr.Length == 0 ? WriteNull(seq) : WriteBytes(seq, bArr),
                 DateTime d => d == DateTime.MinValue ? WriteNull(seq) : WriteTimestamp(seq, d),
-                _ => throw new AMQP.AmqpParseException($"WriteAny Invalid type {value}")
+                _ => throw new AmqpParseException($"WriteAny Invalid type {value}")
             };
-
-            ;
         }
 
-        public static int WriteString(Span<byte> seq, string value)
+        private static int WriteString(Span<byte> seq, string value)
         {
             var len = value.Length;
             var offset = 0;
@@ -55,7 +53,7 @@ public static int WriteString(Span<byte> seq, string value)
             return offset;
         }
 
-        public static int WriteUInt64(Span<byte> seq, ulong value)
+        private static int WriteUInt64(Span<byte> seq, ulong value)
         {
             if (value == 0)
             {
@@ -201,7 +199,7 @@ private static int WriteBool(Span<byte> seq, bool value)
             return WireFormatting.WriteByte(seq, value ? FormatCode.BoolTrue : FormatCode.BoolFalse);
         }
 
-        public static int WriteTimestamp(Span<byte> seq, DateTime value)
+        private static int WriteTimestamp(Span<byte> seq, DateTime value)
         {
             var offset = WireFormatting.WriteByte(seq, FormatCode.Timestamp);
             var unixTime = ((DateTimeOffset)value).ToUnixTimeMilliseconds();
@@ -336,7 +334,7 @@ public static int GetAnySize(object value)
                 byte[] bArr => GetBytesSize(bArr),
                 byte => 1,
                 DateTime d => GetTimestampSize(d),
-                _ => throw new AMQP.AmqpParseException($"WriteAny Invalid type {value}")
+                _ => throw new AmqpParseException($"WriteAny Invalid type {value}")
             };
         }
     }

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -1,3 +1,6 @@
+abstract RabbitMQ.Stream.Client.Reliable.ReliableBase.Close() -> System.Threading.Tasks.Task
+abstract RabbitMQ.Stream.Client.Reliable.ReliableBase.CloseReliable() -> System.Threading.Tasks.Task
+abstract RabbitMQ.Stream.Client.Reliable.ReliableBase.GetNewReliable(bool boot) -> System.Threading.Tasks.Task
 const RabbitMQ.Stream.Client.AMQP.DescribedFormatCode.AmqpValue = 119 -> byte
 const RabbitMQ.Stream.Client.AMQP.DescribedFormatCode.ApplicationData = 117 -> byte
 const RabbitMQ.Stream.Client.AMQP.DescribedFormatCode.ApplicationProperties = 116 -> byte
@@ -84,6 +87,10 @@ const RabbitMQ.Stream.Client.UnsubscribeRequest.Key = 12 -> ushort
 const RabbitMQ.Stream.Client.UnsubscribeResponse.Key = 12 -> ushort
 override RabbitMQ.Stream.Client.LeaderLocator.ToString() -> string
 override RabbitMQ.Stream.Client.LogEventListener.Dispose() -> void
+override RabbitMQ.Stream.Client.Reliable.ReliableConsumer.Close() -> System.Threading.Tasks.Task
+override RabbitMQ.Stream.Client.Reliable.ReliableConsumer.ToString() -> string
+override RabbitMQ.Stream.Client.Reliable.ReliableProducer.Close() -> System.Threading.Tasks.Task
+override RabbitMQ.Stream.Client.Reliable.ReliableProducer.ToString() -> string
 RabbitMQ.Stream.Client.AbstractEntity
 RabbitMQ.Stream.Client.AbstractEntity.AbstractEntity() -> void
 RabbitMQ.Stream.Client.AbstractEntity.client -> RabbitMQ.Stream.Client.Client
@@ -613,18 +620,39 @@ RabbitMQ.Stream.Client.Reliable.IPublishingIdStrategy
 RabbitMQ.Stream.Client.Reliable.IPublishingIdStrategy.GetPublishingId() -> ulong
 RabbitMQ.Stream.Client.Reliable.IPublishingIdStrategy.InitPublishingId() -> System.Threading.Tasks.Task
 RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
-RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenConnected() -> void
-RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenDisconnected(out bool reconnect) -> void
+RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenConnected(string connectionInfo) -> void
+RabbitMQ.Stream.Client.Reliable.IReconnectStrategy.WhenDisconnected(string connectionInfo) -> bool
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.InsertDateTime.get -> System.DateTime
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.InsertDateTime.init -> void
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.Messages.get -> System.Collections.Generic.List<RabbitMQ.Stream.Client.Message>
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.MessagesConfirmation() -> void
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.PublishingId.get -> ulong
 RabbitMQ.Stream.Client.Reliable.MessagesConfirmation.Status.get -> RabbitMQ.Stream.Client.Reliable.ConfirmationStatus
+RabbitMQ.Stream.Client.Reliable.ReliableBase
+RabbitMQ.Stream.Client.Reliable.ReliableBase.Init() -> System.Threading.Tasks.Task
+RabbitMQ.Stream.Client.Reliable.ReliableBase.IsOpen() -> bool
+RabbitMQ.Stream.Client.Reliable.ReliableBase.ReliableBase() -> void
+RabbitMQ.Stream.Client.Reliable.ReliableBase.TryToReconnect(RabbitMQ.Stream.Client.Reliable.IReconnectStrategy reconnectStrategy) -> System.Threading.Tasks.Task
+RabbitMQ.Stream.Client.Reliable.ReliableBase._inReconnection -> bool
+RabbitMQ.Stream.Client.Reliable.ReliableBase._needReconnect -> bool
+RabbitMQ.Stream.Client.Reliable.ReliableConsumer
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.ClientProvidedName.get -> string
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.ClientProvidedName.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.MessageHandler.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.OffsetSpec.get -> RabbitMQ.Stream.Client.IOffsetType
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.OffsetSpec.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.ReconnectStrategy.get -> RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.ReconnectStrategy.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.Reference.get -> string
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.Reference.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.Stream.get -> string
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.Stream.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.StreamSystem.get -> RabbitMQ.Stream.Client.StreamSystem
+RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig.StreamSystem.set -> void
 RabbitMQ.Stream.Client.Reliable.ReliableProducer
-RabbitMQ.Stream.Client.Reliable.ReliableProducer.Close() -> System.Threading.Tasks.Task
-RabbitMQ.Stream.Client.Reliable.ReliableProducer.IsOpen() -> bool
 RabbitMQ.Stream.Client.Reliable.ReliableProducer.Send(RabbitMQ.Stream.Client.Message message) -> System.Threading.Tasks.ValueTask
 RabbitMQ.Stream.Client.Reliable.ReliableProducer.Send(System.Collections.Generic.List<RabbitMQ.Stream.Client.Message> messages, RabbitMQ.Stream.Client.CompressionType compressionType) -> System.Threading.Tasks.ValueTask
 RabbitMQ.Stream.Client.Reliable.ReliableProducerConfig
@@ -812,12 +840,10 @@ RabbitMQ.Stream.Client.UnsubscribeResponse.Write(System.Span<byte> span) -> int
 RabbitMQ.Stream.Client.Version
 RabbitMQ.Stream.Client.VirtualHostAccessFailureException
 RabbitMQ.Stream.Client.VirtualHostAccessFailureException.VirtualHostAccessFailureException(string s) -> void
+readonly RabbitMQ.Stream.Client.Reliable.ReliableBase.SemaphoreSlim -> System.Threading.SemaphoreSlim
 static RabbitMQ.Stream.Client.AMQP.AmqpWireFormatting.GetAnySize(object value) -> int
 static RabbitMQ.Stream.Client.AMQP.AmqpWireFormatting.GetSequenceSize(System.Buffers.ReadOnlySequence<byte> data) -> int
 static RabbitMQ.Stream.Client.AMQP.AmqpWireFormatting.WriteAny(System.Span<byte> seq, object value) -> int
-static RabbitMQ.Stream.Client.AMQP.AmqpWireFormatting.WriteString(System.Span<byte> seq, string value) -> int
-static RabbitMQ.Stream.Client.AMQP.AmqpWireFormatting.WriteTimestamp(System.Span<byte> seq, System.DateTime value) -> int
-static RabbitMQ.Stream.Client.AMQP.AmqpWireFormatting.WriteUInt64(System.Span<byte> seq, ulong value) -> int
 static RabbitMQ.Stream.Client.AMQP.Data.Parse(System.Buffers.ReadOnlySequence<byte> amqpData, ref int byteRead) -> RabbitMQ.Stream.Client.AMQP.Data
 static RabbitMQ.Stream.Client.AMQP.DescribedFormatCode.Read(System.Buffers.ReadOnlySequence<byte> amqpData) -> byte
 static RabbitMQ.Stream.Client.AMQP.DescribedFormatCode.Write(System.Span<byte> span, byte data) -> int
@@ -837,6 +863,7 @@ static RabbitMQ.Stream.Client.PooledTaskSource<T>.Rent() -> RabbitMQ.Stream.Clie
 static RabbitMQ.Stream.Client.PooledTaskSource<T>.Return(RabbitMQ.Stream.Client.ManualResetValueTaskSource<T> task) -> void
 static RabbitMQ.Stream.Client.Producer.Create(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.ProducerConfig config, RabbitMQ.Stream.Client.StreamInfo metaStreamInfo) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Producer>
 static RabbitMQ.Stream.Client.Publish.Version.get -> byte
+static RabbitMQ.Stream.Client.Reliable.ReliableConsumer.CreateReliableConsumer(RabbitMQ.Stream.Client.Reliable.ReliableConsumerConfig reliableConsumerConfig) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Reliable.ReliableConsumer>
 static RabbitMQ.Stream.Client.Reliable.ReliableProducer.CreateReliableProducer(RabbitMQ.Stream.Client.Reliable.ReliableProducerConfig reliableProducerConfig) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.Reliable.ReliableProducer>
 static RabbitMQ.Stream.Client.RoutingHelper<T>.LookupLeaderConnection(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.StreamInfo metaDataInfo) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IClient>
 static RabbitMQ.Stream.Client.RoutingHelper<T>.LookupRandomConnection(RabbitMQ.Stream.Client.ClientParameters clientParameters, RabbitMQ.Stream.Client.StreamInfo metaDataInfo) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IClient>

RabbitMQ.Stream.Client/Reliable/IReconnectStrategy.cs

@@ -2,10 +2,51 @@
 // 2.0, and the Mozilla Public License, version 2.0.
 // Copyright (c) 2007-2020 VMware, Inc.
 
+using System;
+using System.Threading;
+
 namespace RabbitMQ.Stream.Client.Reliable;
 
+/// <summary>
+/// IReconnectStrategy is the interface to reconnect the TCP client
+/// </summary>
 public interface IReconnectStrategy
 {
-    void WhenDisconnected(out bool reconnect);
-    void WhenConnected();
+    /// <summary>
+    /// WhenDisconnected is raised when the TPC client
+    /// is disconnected for some reason. 
+    /// </summary>
+    /// <param name="connectionInfo">Additional connection info. Just for logging</param>
+    /// <returns>if True the client will be reconnected else closed</returns>
+    bool WhenDisconnected(string connectionInfo);
+
+    /// <summary>
+    /// It is raised when the TCP client is connected successfully 
+    /// </summary>
+    /// <param name="connectionInfo">Additional connection info. Just for logging</param>
+    void WhenConnected(string connectionInfo);
+}
+
+/// <summary>
+/// BackOffReconnectStrategy is the default IReconnectStrategy
+/// implementation for Producer and Consumer
+/// It implements a BackOff pattern.
+/// </summary>
+internal class BackOffReconnectStrategy : IReconnectStrategy
+{
+    private int Tentatives { get; set; } = 1;
+
+    public bool WhenDisconnected(string connectionInfo)
+    {
+        Tentatives <<= 1;
+        LogEventSource.Log.LogInformation(
+            $"{connectionInfo} disconnected, check if reconnection needed in {Tentatives * 100} ms.");
+        Thread.Sleep(TimeSpan.FromMilliseconds(Tentatives * 100));
+        return true;
+    }
+
+    public void WhenConnected(string _)
+    {
+        Tentatives = 1;
+    }
 }