Improves shutdown and cleanup logic for producers and consumers (#440)

This PR improves shutdown and cleanup logic for producers and consumers, focusing on graceful termination and better error handling during close operations.

- Adds cancellation token checks to prevent processing loops from continuing after shutdown
- Ensures channel writers are properly completed to unblock waiting consumers
- Demotes timeout error logs to debug level to reduce log noise during normal shutdown scenarios
- Consolidates ignore logic for cleaner handling of already-closed connections
---------

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/AbstractEntity.cs

@@ -93,12 +93,10 @@ protected async Task<ResponseCode> Shutdown(EntityCommonConfig config, bool igno
             }
 
             UpdateStatusToClosed();
-            var result = await DeleteEntityFromTheServer(ignoreIfAlreadyDeleted).ConfigureAwait(false);
-
-            if (_client is { IsClosed: true })
-            {
-                return result;
-            }
+            // we can ignore if the ignoreIfAlreadyDeleted or the socket is already closed
+            // DeleteEntityFromTheServer must be called anyway because it cleans the internal lists
+            var ignore = ignoreIfAlreadyDeleted || _client.IsClosed;
+            var result = await DeleteEntityFromTheServer(ignore).ConfigureAwait(false);
 
             var closed = await _client.MaybeClose($"closing: {EntityId}", config.Pool)
                 .ConfigureAwait(false);

RabbitMQ.Stream.Client/Client.cs

@@ -772,7 +772,7 @@ private async Task<CloseResponse> Close(string reason, string closedStatus)
             }
             catch (TimeoutException)
             {
-                _logger.LogError(
+                _logger.LogDebug(
                     "Timeout while closing the connection. The connection will be closed anyway");
             }
             catch (Exception e)

RabbitMQ.Stream.Client/RawConsumer.cs

@@ -457,7 +457,7 @@ private void ProcessChunks()
             {
                 // need to wait the subscription is completed 
                 // else the _subscriberId could be incorrect
-                _completeSubscription.Task.Wait();
+                await _completeSubscription.Task.ConfigureAwait(false);
 
                 try
                 {
@@ -555,6 +555,18 @@ await _client.Credit(EntityId, 1)
                         "Error while process chunks the stream: {EntityInfo} The ProcessChunks task will be closed",
                         DumpEntityConfiguration());
                 }
+                finally
+                {
+                    // best-effort: mark the writer complete so no producers stay blocked
+                    try
+                    {
+                        _chunksBuffer.Writer.TryComplete();
+                    }
+                    catch
+                    {
+                        // ignored
+                    }
+                }
             }, Token);
         }
 
@@ -764,7 +776,7 @@ protected override async Task<ResponseCode> DeleteEntityFromTheServer(bool ignor
 
             catch (TimeoutException)
             {
-                Logger.LogError(
+                Logger.LogDebug(
                     "Timeout removing the consumer id: {SubscriberId}, {EntityInfo} from the server. " +
                     "The consumer will be closed anyway",
                     EntityId, DumpEntityConfiguration());
@@ -785,7 +797,7 @@ public override async Task<ResponseCode> Close()
             // when the consumer is closed we must be sure that the 
             // subscription is completed to avoid problems with the connection
             // It could happen when the closing is called just after the creation
-            _completeSubscription.Task.Wait();
+            await _completeSubscription.Task.ConfigureAwait(false);
             return await Shutdown(_config).ConfigureAwait(false);
         }
 

RabbitMQ.Stream.Client/RawProducer.cs

@@ -346,7 +346,7 @@ private async Task ProcessBuffer()
             try
             {
                 var messages = new List<(ulong, Message)>(_config.MessagesBufferSize);
-                while (await _messageBuffer.Reader.WaitToReadAsync().ConfigureAwait(false))
+                while (await _messageBuffer.Reader.WaitToReadAsync(Token).ConfigureAwait(false))
                 {
                     while (_messageBuffer.Reader.TryRead(out var msg))
                     {
@@ -371,6 +371,16 @@ private async Task ProcessBuffer()
 
         public override async Task<ResponseCode> Close()
         {
+            // Complete the channel writer to signal ProcessBuffer to finish
+            try
+            {
+                _messageBuffer.Writer.TryComplete();
+            }
+            catch
+            {
+                // ignored
+            }
+
             return await Shutdown(_config).ConfigureAwait(false);
         }
 
@@ -382,8 +392,11 @@ protected override async Task<ResponseCode> DeleteEntityFromTheServer(bool ignor
                 // in this case we reduce the waiting time
                 // the producer could be removed because of stream deleted 
                 // so it is not necessary to wait.
+                // we can ignore if the ignoreIfAlreadyDeleted or the socket is already closed
+                // DeletePublisher must be called anyway because it cleans the internal lists
+                var ignore = ignoreIfAlreadyDeleted || _client.IsClosed;
                 var closeResponse =
-                    await _client.DeletePublisher(EntityId, ignoreIfAlreadyDeleted).ConfigureAwait(false);
+                    await _client.DeletePublisher(EntityId, ignore).ConfigureAwait(false);
                 return closeResponse.ResponseCode;
             }
             catch (Exception e)