Fixed forwarding of StatusChanged event (#432)

* Fixed forwarding of StatusChanged event

* add tests

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

---------

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

Author: BlindOP

RabbitMQ.Stream.Client/Reliable/DeduplicatingProducer.cs

@@ -39,22 +39,7 @@ public static async Task<DeduplicatingProducer> Create(DeduplicatingProducerConf
         var x = new DeduplicatingProducer()
         {
             _producer = await Producer
-                .Create(
-                    new ProducerConfig(producerConfig.StreamSystem, producerConfig.Stream)
-                    {
-                        _reference = producerConfig.Reference,
-                        ConfirmationHandler = producerConfig.ConfirmationHandler,
-                        ReconnectStrategy = producerConfig.ReconnectStrategy,
-                        ClientProvidedName = producerConfig.ClientProvidedName,
-                        SuperStreamConfig = producerConfig.SuperStreamConfig,
-                        MaxInFlight = producerConfig.MaxInFlight,
-                        MessagesBufferSize = producerConfig.MessagesBufferSize,
-                        TimeoutMessageAfter = producerConfig.TimeoutMessageAfter,
-                        Filter = producerConfig.Filter,
-                        Identifier = producerConfig.Identifier,
-                        ResourceAvailableReconnectStrategy = producerConfig.ResourceAvailableReconnectStrategy,
-
-                    }, logger)
+                .Create(producerConfig, logger)
                 .ConfigureAwait(false)
         };
         return x;

Tests/DeduplicationProducerTests.cs

@@ -3,6 +3,7 @@
 // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 
 using System;
+using System.Collections.Generic;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -80,16 +81,22 @@ public async Task DeduplicationInActionSendingTheSameIdMessagesWontStore()
         SystemUtils.InitStreamSystemWithRandomStream(out var system, out var stream);
         var testPassed = new TaskCompletionSource<ulong>();
         const ulong TotalMessages = 1000UL;
-        var p = await DeduplicatingProducer.Create(
-            new DeduplicatingProducerConfig(system, stream, "my_producer_reference")
+        var dupConfig = new DeduplicatingProducerConfig(system, stream, "my_producer_reference")
+        {
+            ConfirmationHandler = async confirmation =>
             {
-                ConfirmationHandler = async confirmation =>
-                {
-                    if (confirmation.PublishingId == TotalMessages)
-                        testPassed.SetResult(TotalMessages);
-                    await Task.CompletedTask;
-                },
-            });
+                if (confirmation.PublishingId == TotalMessages)
+                    testPassed.SetResult(TotalMessages);
+                await Task.CompletedTask;
+            },
+        };
+        var statusInfoReceived = new List<StatusInfo>();
+        dupConfig.StatusChanged += (status) => { statusInfoReceived.Add(status); };
+
+        var p = await DeduplicatingProducer.Create(dupConfig);
+
+        Assert.Equal(ReliableEntityStatus.Initialization, statusInfoReceived[0].From);
+        Assert.Equal(ReliableEntityStatus.Open, statusInfoReceived[0].To);
         // first send and the messages are stored
         for (ulong i = 1; i <= TotalMessages; i++)
         {