Add event status changed (#349)

Add StatusChangedHandler StatusChanged to the Producer and Consumer classes. It exposes the internal status with `From` and `To` and the reason for the status change. The user is aware of what is happening internally and makes decisions based on the internal status. The Producer and Consumer classes are considered `Open` even if the status is: `Reconnecting`. --------- Signed-off-by: Gabriele Santomaggio <[email protected]>
rabbitmq · Jan 30, 2024 · d70458a · d70458a
1 parent 85c220a
commit d70458a
Show file tree

Hide file tree

Showing 16 changed files with 812 additions and 123 deletions.
diff --git a/Directory.Packages.props b/Directory.Packages.props
@@ -21,18 +21,16 @@
     <PackageVersion Include="Microsoft.Extensions.Logging" Version="7.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Console" Version="7.0.0" />
   </ItemGroup>
-
   <ItemGroup Label=".NET 6 Specific" Condition="'$(TargetFramework)' == 'net6.0'">
     <!-- RabbitMQ.Stream.Client -->
     <PackageVersion Include="System.IO.Hashing" Version="6.0.0" />
     <PackageVersion Include="System.IO.Pipelines" Version="6.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.0" />
   </ItemGroup>
-
   <ItemGroup Label=".NET 7 Specific" Condition="'$(TargetFramework)' == 'net7.0'">
     <!-- RabbitMQ.Stream.Client -->
     <PackageVersion Include="System.IO.Hashing" Version="7.0.0" />
     <PackageVersion Include="System.IO.Pipelines" Version="7.0.0" />
     <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="7.0.0" />
-  </ItemGroup>  
-</Project>
+  </ItemGroup>
+</Project>
diff --git a/RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt b/RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt
@@ -184,6 +184,13 @@ RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.ConnectionClosedHandler.get
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.ConnectionClosedHandler.set -> void
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.get -> RabbitMQ.Stream.Client.RoutingStrategyType
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.set -> void
+RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
+RabbitMQ.Stream.Client.Reliable.ChangeStatusReason.BoolFailure = 5 -> RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
+RabbitMQ.Stream.Client.Reliable.ChangeStatusReason.ClosedByStrategyPolicy = 4 -> RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
+RabbitMQ.Stream.Client.Reliable.ChangeStatusReason.ClosedByUser = 3 -> RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
+RabbitMQ.Stream.Client.Reliable.ChangeStatusReason.MetaDataUpdate = 2 -> RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
+RabbitMQ.Stream.Client.Reliable.ChangeStatusReason.None = 0 -> RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
+RabbitMQ.Stream.Client.Reliable.ChangeStatusReason.UnexpectedlyDisconnected = 1 -> RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
 RabbitMQ.Stream.Client.Reliable.Consumer.Info.get -> RabbitMQ.Stream.Client.ConsumerInfo
 RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Crc32.get -> RabbitMQ.Stream.Client.ICrc32
 RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Crc32.set -> void
@@ -209,18 +216,35 @@ RabbitMQ.Stream.Client.Reliable.ProducerConfig.Filter.set -> void
 RabbitMQ.Stream.Client.Reliable.ProducerConfig.Reference.set -> void
 RabbitMQ.Stream.Client.Reliable.ProducerFactory.CreateProducer(bool boot) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.IProducer>
 RabbitMQ.Stream.Client.Reliable.ProducerFactory._producer -> RabbitMQ.Stream.Client.IProducer
-RabbitMQ.Stream.Client.Reliable.ReliableBase.UpdateStatus(RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus status) -> void
+RabbitMQ.Stream.Client.Reliable.ReliableBase.UpdateStatus(RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus newStatus, RabbitMQ.Stream.Client.Reliable.ChangeStatusReason reason, string partition = null) -> void
 RabbitMQ.Stream.Client.Reliable.ReliableBase._status -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
 RabbitMQ.Stream.Client.Reliable.ReliableConfig.Identifier.get -> string
 RabbitMQ.Stream.Client.Reliable.ReliableConfig.Identifier.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConfig.OnStatusChanged(RabbitMQ.Stream.Client.Reliable.StatusInfo statusInfo) -> void
 RabbitMQ.Stream.Client.Reliable.ReliableConfig.ReconnectStrategy.get -> RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
 RabbitMQ.Stream.Client.Reliable.ReliableConfig.ResourceAvailableReconnectStrategy.get -> RabbitMQ.Stream.Client.Reliable.IReconnectStrategy
 RabbitMQ.Stream.Client.Reliable.ReliableConfig.ResourceAvailableReconnectStrategy.set -> void
+RabbitMQ.Stream.Client.Reliable.ReliableConfig.StatusChanged -> RabbitMQ.Stream.Client.Reliable.ReliableConfig.StatusChangedHandler
+RabbitMQ.Stream.Client.Reliable.ReliableConfig.StatusChangedHandler
 RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
 RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus.Closed = 3 -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
 RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus.Initialization = 0 -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
 RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus.Open = 1 -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
-RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus.Reconnecting = 2 -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
+RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus.Reconnection = 2 -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
+RabbitMQ.Stream.Client.Reliable.StatusInfo
+RabbitMQ.Stream.Client.Reliable.StatusInfo.From.get -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
+RabbitMQ.Stream.Client.Reliable.StatusInfo.From.init -> void
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Identifier.get -> string
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Identifier.init -> void
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Partition.get -> string
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Partition.init -> void
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Reason.get -> RabbitMQ.Stream.Client.Reliable.ChangeStatusReason
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Reason.init -> void
+RabbitMQ.Stream.Client.Reliable.StatusInfo.StatusInfo(RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus From, RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus To, string Stream, string Identifier, string Partition, RabbitMQ.Stream.Client.Reliable.ChangeStatusReason Reason = RabbitMQ.Stream.Client.Reliable.ChangeStatusReason.None) -> void
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Stream.get -> string
+RabbitMQ.Stream.Client.Reliable.StatusInfo.Stream.init -> void
+RabbitMQ.Stream.Client.Reliable.StatusInfo.To.get -> RabbitMQ.Stream.Client.Reliable.ReliableEntityStatus
+RabbitMQ.Stream.Client.Reliable.StatusInfo.To.init -> void
 RabbitMQ.Stream.Client.Reliable.SuperStreamConfig.RoutingStrategyType.get -> RabbitMQ.Stream.Client.RoutingStrategyType
 RabbitMQ.Stream.Client.Reliable.SuperStreamConfig.RoutingStrategyType.set -> void
 RabbitMQ.Stream.Client.RouteNotFoundException

diff --git a/RabbitMQ.Stream.Client/Reliable/Consumer.cs b/RabbitMQ.Stream.Client/Reliable/Consumer.cs
@@ -172,7 +172,7 @@ public static async Task<Consumer> Create(ConsumerConfig consumerConfig, ILogger
         consumerConfig.ReconnectStrategy ??= new BackOffReconnectStrategy(logger);
         consumerConfig.ResourceAvailableReconnectStrategy ??= new ResourceAvailableBackOffReconnectStrategy(logger);
         var rConsumer = new Consumer(consumerConfig, logger);
-        await rConsumer.Init(consumerConfig.ReconnectStrategy, consumerConfig.ResourceAvailableReconnectStrategy)
+        await rConsumer.Init(consumerConfig)
             .ConfigureAwait(false);
         return rConsumer;
     }
@@ -204,11 +204,11 @@ public override async Task Close()
     {
         if (_status == ReliableEntityStatus.Initialization)
         {
-            UpdateStatus(ReliableEntityStatus.Closed);
+            UpdateStatus(ReliableEntityStatus.Closed, ChangeStatusReason.ClosedByUser);
             return;
         }
 
-        UpdateStatus(ReliableEntityStatus.Closed);
+        UpdateStatus(ReliableEntityStatus.Closed, ChangeStatusReason.ClosedByUser);
         await CloseEntity().ConfigureAwait(false);
         _logger?.LogDebug("Consumer {Identity} closed", ToString());
     }

diff --git a/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs b/RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs
@@ -62,15 +62,18 @@ private async Task<IConsumer> StandardConsumer(bool boot)
             {
                 if (closeReason == ConnectionClosedReason.Normal)
                 {
+                    //  we don't update the status here since it happens when Close() is called in a normal way
                     BaseLogger.LogInformation("{Identity} is closed normally", ToString());
                     return;
                 }
 
-                await OnEntityClosed(_consumerConfig.StreamSystem, _consumerConfig.Stream).ConfigureAwait(false);
+                await OnEntityClosed(_consumerConfig.StreamSystem, _consumerConfig.Stream,
+                    ChangeStatusReason.UnexpectedlyDisconnected).ConfigureAwait(false);
             },
             MetadataHandler = async _ =>
             {
-                await OnEntityClosed(_consumerConfig.StreamSystem, _consumerConfig.Stream).ConfigureAwait(false);
+                await OnEntityClosed(_consumerConfig.StreamSystem, _consumerConfig.Stream,
+                    ChangeStatusReason.MetaDataUpdate).ConfigureAwait(false);
             },
             MessageHandler = async (consumer, ctx, message) =>
             {
@@ -127,19 +130,22 @@ private async Task<IConsumer> SuperConsumer(bool boot)
                         await RandomWait().ConfigureAwait(false);
                         if (closeReason == ConnectionClosedReason.Normal)
                         {
+                            //  we don't update the status here since it happens when Close() is called in a normal way
                             BaseLogger.LogInformation("{Identity} is closed normally", ToString());
                             return;
                         }
 
                         var r = ((RawSuperStreamConsumer)(_consumer)).ReconnectPartition;
-                        await OnEntityClosed(_consumerConfig.StreamSystem, partitionStream, r)
+                        await OnEntityClosed(_consumerConfig.StreamSystem, partitionStream, r,
+                                ChangeStatusReason.UnexpectedlyDisconnected)
                             .ConfigureAwait(false);
                     },
                     MetadataHandler = async update =>
                     {
                         await RandomWait().ConfigureAwait(false);
                         var r = ((RawSuperStreamConsumer)(_consumer)).ReconnectPartition;
-                        await OnEntityClosed(_consumerConfig.StreamSystem, update.Stream, r)
+                        await OnEntityClosed(_consumerConfig.StreamSystem, update.Stream, r,
+                                ChangeStatusReason.MetaDataUpdate)
                             .ConfigureAwait(false);
                     },
                     MessageHandler = async (partitionStream, consumer, ctx, message) =>

diff --git a/RabbitMQ.Stream.Client/Reliable/Producer.cs b/RabbitMQ.Stream.Client/Reliable/Producer.cs
@@ -159,7 +159,7 @@ public static async Task<Producer> Create(ProducerConfig producerConfig, ILogger
         producerConfig.ReconnectStrategy ??= new BackOffReconnectStrategy(logger);
         producerConfig.ResourceAvailableReconnectStrategy ??= new ResourceAvailableBackOffReconnectStrategy(logger);
         var rProducer = new Producer(producerConfig, logger);
-        await rProducer.Init(producerConfig.ReconnectStrategy, producerConfig.ResourceAvailableReconnectStrategy)
+        await rProducer.Init(producerConfig)
             .ConfigureAwait(false);
         return rProducer;
     }
@@ -201,11 +201,11 @@ public override async Task Close()
     {
         if (ReliableEntityStatus.Initialization == _status)
         {
-            UpdateStatus(ReliableEntityStatus.Closed);
+            UpdateStatus(ReliableEntityStatus.Closed, ChangeStatusReason.ClosedByUser);
             return;
         }
 
-        UpdateStatus(ReliableEntityStatus.Closed);
+        UpdateStatus(ReliableEntityStatus.Closed, ChangeStatusReason.ClosedByUser);
         await SemaphoreSlim.WaitAsync().ConfigureAwait(false);
         try
         {

diff --git a/RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs b/RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs
@@ -34,7 +34,6 @@ private async Task<IProducer> SuperStreamProducer(bool boot)
     {
         if (boot)
         {
-
             return await _producerConfig.StreamSystem.CreateRawSuperStreamProducer(
                 new RawSuperStreamProducerConfig(_producerConfig.Stream)
                 {
@@ -56,14 +55,16 @@ private async Task<IProducer> SuperStreamProducer(bool boot)
                         }
 
                         var r = ((RawSuperStreamProducer)(_producer)).ReconnectPartition;
-                        await OnEntityClosed(_producerConfig.StreamSystem, partitionStream, r)
+                        await OnEntityClosed(_producerConfig.StreamSystem, partitionStream, r,
+                                ChangeStatusReason.UnexpectedlyDisconnected)
                             .ConfigureAwait(false);
                     },
                     MetadataHandler = async update =>
                     {
                         await RandomWait().ConfigureAwait(false);
                         var r = ((RawSuperStreamProducer)(_producer)).ReconnectPartition;
-                        await OnEntityClosed(_producerConfig.StreamSystem, update.Stream, r)
+                        await OnEntityClosed(_producerConfig.StreamSystem, update.Stream, r,
+                                ChangeStatusReason.MetaDataUpdate)
                             .ConfigureAwait(false);
                     },
                     ConfirmHandler = confirmationHandler =>
@@ -104,7 +105,8 @@ private async Task<IProducer> StandardProducer()
             MetadataHandler = async _ =>
             {
                 await RandomWait().ConfigureAwait(false);
-                await OnEntityClosed(_producerConfig.StreamSystem, _producerConfig.Stream).ConfigureAwait(false);
+                await OnEntityClosed(_producerConfig.StreamSystem, _producerConfig.Stream,
+                    ChangeStatusReason.MetaDataUpdate).ConfigureAwait(false);
             },
             ConnectionClosedHandler = async (closeReason) =>
             {
@@ -115,7 +117,8 @@ private async Task<IProducer> StandardProducer()
                     return;
                 }
 
-                await OnEntityClosed(_producerConfig.StreamSystem, _producerConfig.Stream).ConfigureAwait(false);
+                await OnEntityClosed(_producerConfig.StreamSystem, _producerConfig.Stream,
+                    ChangeStatusReason.UnexpectedlyDisconnected).ConfigureAwait(false);
             },
             ConfirmHandler = confirmation =>
             {