rabbitmq · Gsantomaggio · Jan 30, 2024 · Jan 26, 2024 · Jan 27, 2024 · Jan 29, 2024
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 =>
             {