Refactor RawProducer and RawConsumer classes

Remove duplication code between the Raw* classes and moved to
AbstractEntity class.

Fix a rare condition where the producer still receive confirmation but
the producer was removed from the internal list due of cancellation.

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/AbstractEntity.cs

@@ -9,6 +9,12 @@
 
 namespace RabbitMQ.Stream.Client
 {
+
+    public abstract record EntityCommonConfig
+    {
+        internal ConnectionsPool Pool { get; set; }
+    }
+
     internal enum EntityStatus
     {
         Open,
@@ -18,28 +24,73 @@ internal enum EntityStatus
 
     public interface IClosable
     {
-        public Task<ResponseCode> Close(bool ignoreIfClosed = false);
+        public Task<ResponseCode> Close();
     }
 
     public abstract class AbstractEntity : IClosable
     {
         private readonly CancellationTokenSource _cancelTokenSource = new();
         protected CancellationToken Token => _cancelTokenSource.Token;
-
+        protected ILogger Logger { get; init; }
         internal EntityStatus _status = EntityStatus.Closed;
 
+        protected byte EntityId { get; set; }
+        protected abstract string GetStream();
+        protected abstract string DumpEntityConfiguration();
+
         // here the _cancelTokenSource is disposed and the token is cancelled
         // in producer is used to cancel the send task
         // in consumer is used to cancel the receive task
-        protected void MaybeCancelToken()
+        private void MaybeCancelToken()
         {
             if (!_cancelTokenSource.IsCancellationRequested)
                 _cancelTokenSource.Cancel();
         }
 
-        public abstract Task<ResponseCode> Close(bool ignoreIfClosed = false);
+        public abstract Task<ResponseCode> Close();
+
+        /// <summary>
+        /// Remove the producer or consumer from the server
+        /// </summary>
+        /// <param name="ignoreIfAlreadyDeleted"> In case the producer or consumer is already removed from the server.
+        /// ex: metadata update </param>
+        /// <returns></returns>
+        protected abstract Task<ResponseCode> DeleteEntityFromTheServer(bool ignoreIfAlreadyDeleted = false);
+
+        /// <summary>
+        /// Internal close method. It is called by the public Close method.
+        /// Set the status to closed and remove the producer or consumer from the server ( if it is not already removed )
+        /// Close the TCP connection if it is not already closed or it is needed.
+        /// </summary>
+        /// <param name="config">The connection pool instance</param>
+        /// <param name="ignoreIfAlreadyDeleted"></param>
+        /// <returns></returns>
+        protected async Task<ResponseCode> Shutdown(EntityCommonConfig config, bool ignoreIfAlreadyDeleted = false)
+        {
+            MaybeCancelToken();
+
+            if (!IsOpen()) // the client is already closed
+            {
+                return ResponseCode.Ok;
+            }
+
+            _status = EntityStatus.Closed;
+            var result = await DeleteEntityFromTheServer(ignoreIfAlreadyDeleted).ConfigureAwait(false);
+
+            if (_client is { IsClosed: true })
+            {
+                return result;
+            }
+
+            var closed = await _client.MaybeClose($"closing: {EntityId}",
+                    GetStream(), config.Pool)
+                .ConfigureAwait(false);
+            ClientExceptions.MaybeThrowException(closed.ResponseCode, $"_client-close-Entity: {EntityId}");
+            Logger.LogDebug("{EntityInfo} is closed", DumpEntityConfiguration());
+            return result;
+        }
 
-        protected void Dispose(bool disposing, string entityInfo, ILogger logger)
+        protected void Dispose(bool disposing)
         {
             if (!disposing)
             {
@@ -56,12 +107,12 @@ protected void Dispose(bool disposing, string entityInfo, ILogger logger)
                 var closeTask = Close();
                 if (!closeTask.Wait(Consts.MidWait))
                 {
-                    logger.LogWarning("Failed to close {EntityInfo} in time", entityInfo);
+                    Logger?.LogWarning("Failed to close {EntityInfo} in time", DumpEntityConfiguration());
                 }
             }
             catch (Exception e)
             {
-                logger?.LogWarning("Failed to close {EntityInfo}, error {Error} ", entityInfo, e.Message);
+                Logger?.LogWarning("Failed to close {EntityInfo}, error {Error} ", DumpEntityConfiguration(), e.Message);
             }
             finally
             {

RabbitMQ.Stream.Client/Client.cs

@@ -331,22 +331,28 @@ public ValueTask<bool> Publish<T>(T msg) where T : struct, ICommand
             return (publisherId, response);
         }
 
-        public async Task<DeletePublisherResponse> DeletePublisher(byte publisherId)
+        public async Task<DeletePublisherResponse> DeletePublisher(byte publisherId,
+            bool ignoreIfAlreadyRemoved = false)
         {
             await _poolSemaphore.WaitAsync().ConfigureAwait(false);
             try
             {
-                var result =
-                    await Request<DeletePublisherRequest, DeletePublisherResponse>(corr =>
-                        new DeletePublisherRequest(corr, publisherId)).ConfigureAwait(false);
+                if (!ignoreIfAlreadyRemoved)
+                {
+                    var result =
+                        await Request<DeletePublisherRequest, DeletePublisherResponse>(corr =>
+                            new DeletePublisherRequest(corr, publisherId)).ConfigureAwait(false);
 
-                return result;
+                    return result;
+                }
             }
             finally
             {
                 publishers.Remove(publisherId);
                 _poolSemaphore.Release();
             }
+
+            return new DeletePublisherResponse();
         }
 
         public async Task<(byte, SubscribeResponse)> Subscribe(string stream, IOffsetType offsetType,
@@ -492,12 +498,25 @@ private async Task HandleIncoming(Memory<byte> frameMemory)
                 case PublishConfirm.Key:
                     PublishConfirm.Read(frame, out var confirm);
                     confirmFrames += 1;
-                    var (confirmCallback, _) = publishers[confirm.PublisherId];
-                    confirmCallback(confirm.PublishingIds);
-                    if (MemoryMarshal.TryGetArray(confirm.PublishingIds, out var confirmSegment))
+                    if (publishers.TryGetValue(confirm.PublisherId, out var publisherConf))
                     {
-                        if (confirmSegment.Array != null)
-                            ArrayPool<ulong>.Shared.Return(confirmSegment.Array);
+                        var (confirmCallback, _) = publisherConf;
+                        confirmCallback(confirm.PublishingIds);
+                        if (MemoryMarshal.TryGetArray(confirm.PublishingIds, out var confirmSegment))
+                        {
+                            if (confirmSegment.Array != null)
+                                ArrayPool<ulong>.Shared.Return(confirmSegment.Array);
+                        }
+                    }
+                    else
+                    {
+                        // the producer is not found, this can happen when the producer is closing
+                        // and there are still confirmation on the wire 
+                        // we can ignore the error since the producer does not exists anymore
+                        _logger?.LogDebug(
+                            "Could not find stream producer {ID} or producer is closing." +
+                            "A possible cause it that the producer was closed and the are still confirmation on the wire. ",
+                            confirm.PublisherId);
                     }
 
                     break;
@@ -522,8 +541,22 @@ private async Task HandleIncoming(Memory<byte> frameMemory)
                     break;
                 case PublishError.Key:
                     PublishError.Read(frame, out var error);
-                    var (_, errorCallback) = publishers[error.PublisherId];
-                    errorCallback(error.PublishingErrors);
+                    if (publishers.TryGetValue(error.PublisherId, out var publisher))
+                    {
+                        var (_, errorCallback) = publisher;
+                        errorCallback(error.PublishingErrors);
+                    }
+                    else
+                    {
+                        // the producer is not found, this can happen when the producer is closing
+                        // and there are still confirmation on the wire 
+                        // we can ignore the error since the producer does not exists anymore
+                        _logger?.LogDebug(
+                            "Could not find stream producer {ID} or producer is closing." +
+                            "A possible cause it that the producer was closed and the are still confirmation on the wire. ",
+                            error.PublisherId);
+                    }
+
                     break;
                 case MetaDataUpdate.Key:
                     MetaDataUpdate.Read(frame, out var metaDataUpdate);

RabbitMQ.Stream.Client/IConsumer.cs

@@ -15,12 +15,10 @@ public interface IConsumer : IClosable
     public ConsumerInfo Info { get; }
 }
 
-public record IConsumerConfig : INamedEntity
+public record IConsumerConfig : EntityCommonConfig, INamedEntity
 {
     private ushort _initialCredits = Consts.ConsumerInitialCredits;
 
-    internal ConnectionsPool Pool { get; set; }
-
     // StoredOffsetSpec configuration it is needed to keep the offset spec.
     // since the offset can be decided from the ConsumerConfig.OffsetSpec.
     // and from ConsumerConfig.ConsumerUpdateListener.

RabbitMQ.Stream.Client/IProducer.cs

@@ -81,11 +81,9 @@ public record ProducerFilter
     public Func<Message, string> FilterValue { get; set; } = null;
 }
 
-public record IProducerConfig : INamedEntity
+public record IProducerConfig : EntityCommonConfig, INamedEntity
 {
 
-    internal ConnectionsPool Pool { get; set; }
-
     public string Reference { get; set; }
     public int MaxInFlight { get; set; } = 1_000;
     public string ClientProvidedName { get; set; } = "dotnet-stream-raw-producer";

RabbitMQ.Stream.Client/PublicAPI.Shipped.txt

@@ -169,7 +169,6 @@ RabbitMQ.Stream.Client.Client.ConnectionCloseHandler
 RabbitMQ.Stream.Client.Client.ConnectionProperties.get -> System.Collections.Generic.IDictionary<string, string>
 RabbitMQ.Stream.Client.Client.CreateStream(string stream, System.Collections.Generic.IDictionary<string, string> args) -> System.Threading.Tasks.ValueTask<RabbitMQ.Stream.Client.CreateResponse>
 RabbitMQ.Stream.Client.Client.Credit(byte subscriptionId, ushort credit) -> System.Threading.Tasks.ValueTask<bool>
-RabbitMQ.Stream.Client.Client.DeletePublisher(byte publisherId) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.DeletePublisherResponse>
 RabbitMQ.Stream.Client.Client.DeleteStream(string stream) -> System.Threading.Tasks.ValueTask<RabbitMQ.Stream.Client.DeleteResponse>
 RabbitMQ.Stream.Client.Client.IncomingFrames.get -> int
 RabbitMQ.Stream.Client.Client.IsClosed.get -> bool

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -1,12 +1,19 @@
-abstract RabbitMQ.Stream.Client.AbstractEntity.Close(bool ignoreIfClosed = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+abstract RabbitMQ.Stream.Client.AbstractEntity.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+abstract RabbitMQ.Stream.Client.AbstractEntity.DeleteEntityFromTheServer(bool ignoreIfAlreadyDeleted = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+abstract RabbitMQ.Stream.Client.AbstractEntity.DumpEntityConfiguration() -> string
+abstract RabbitMQ.Stream.Client.AbstractEntity.GetStream() -> string
 const RabbitMQ.Stream.Client.RouteQueryResponse.Key = 24 -> ushort
 const RabbitMQ.Stream.Client.StreamStatsResponse.Key = 28 -> ushort
 override RabbitMQ.Stream.Client.Broker.ToString() -> string
-override RabbitMQ.Stream.Client.RawConsumer.Close(bool ignoreIfClosed = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
-override RabbitMQ.Stream.Client.RawProducer.Close(bool ignoreIfClosed = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
-RabbitMQ.Stream.Client.AbstractEntity.Dispose(bool disposing, string entityInfo, Microsoft.Extensions.Logging.ILogger logger) -> void
+override RabbitMQ.Stream.Client.RawConsumer.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+override RabbitMQ.Stream.Client.RawProducer.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+RabbitMQ.Stream.Client.AbstractEntity.Dispose(bool disposing) -> void
+RabbitMQ.Stream.Client.AbstractEntity.EntityId.get -> byte
+RabbitMQ.Stream.Client.AbstractEntity.EntityId.set -> void
 RabbitMQ.Stream.Client.AbstractEntity.IsOpen() -> bool
-RabbitMQ.Stream.Client.AbstractEntity.MaybeCancelToken() -> void
+RabbitMQ.Stream.Client.AbstractEntity.Logger.get -> Microsoft.Extensions.Logging.ILogger
+RabbitMQ.Stream.Client.AbstractEntity.Logger.init -> void
+RabbitMQ.Stream.Client.AbstractEntity.Shutdown(RabbitMQ.Stream.Client.EntityCommonConfig config, bool ignoreIfAlreadyDeleted = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
 RabbitMQ.Stream.Client.AbstractEntity.Token.get -> System.Threading.CancellationToken
 RabbitMQ.Stream.Client.AuthMechanism
 RabbitMQ.Stream.Client.AuthMechanism.External = 1 -> RabbitMQ.Stream.Client.AuthMechanism
@@ -19,6 +26,7 @@ RabbitMQ.Stream.Client.Chunk.MagicVersion.get -> byte
 RabbitMQ.Stream.Client.Client.ClientId.get -> string
 RabbitMQ.Stream.Client.Client.ClientId.init -> void
 RabbitMQ.Stream.Client.Client.DeclarePublisher(string publisherRef, string stream, System.Action<System.ReadOnlyMemory<ulong>> confirmCallback, System.Action<(ulong, RabbitMQ.Stream.Client.ResponseCode)[]> errorCallback, RabbitMQ.Stream.Client.ConnectionsPool pool = null) -> System.Threading.Tasks.Task<(byte, RabbitMQ.Stream.Client.DeclarePublisherResponse)>
+RabbitMQ.Stream.Client.Client.DeletePublisher(byte publisherId, bool ignoreIfAlreadyRemoved = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.DeletePublisherResponse>
 RabbitMQ.Stream.Client.Client.ExchangeVersions() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.CommandVersionsResponse>
 RabbitMQ.Stream.Client.Client.QueryRoute(string superStream, string routingKey) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.RouteQueryResponse>
 RabbitMQ.Stream.Client.Client.StreamStats(string stream) -> System.Threading.Tasks.ValueTask<RabbitMQ.Stream.Client.StreamStatsResponse>
@@ -63,11 +71,12 @@ RabbitMQ.Stream.Client.ConsumerInfo.ConsumerInfo(string stream, string reference
 RabbitMQ.Stream.Client.ConsumerInfo.Reference.get -> string
 RabbitMQ.Stream.Client.CrcException
 RabbitMQ.Stream.Client.CrcException.CrcException(string s) -> void
+RabbitMQ.Stream.Client.EntityCommonConfig
 RabbitMQ.Stream.Client.HashRoutingMurmurStrategy.Route(RabbitMQ.Stream.Client.Message message, System.Collections.Generic.List<string> partitions) -> System.Threading.Tasks.Task<System.Collections.Generic.List<string>>
 RabbitMQ.Stream.Client.IClient.ClientId.get -> string
 RabbitMQ.Stream.Client.IClient.ClientId.init -> void
 RabbitMQ.Stream.Client.IClosable
-RabbitMQ.Stream.Client.IClosable.Close(bool ignoreIfClosed = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+RabbitMQ.Stream.Client.IClosable.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
 RabbitMQ.Stream.Client.IConsumer.Info.get -> RabbitMQ.Stream.Client.ConsumerInfo
 RabbitMQ.Stream.Client.IConsumerConfig.Crc32.get -> RabbitMQ.Stream.Client.ICrc32
 RabbitMQ.Stream.Client.IConsumerConfig.Crc32.set -> void
@@ -128,9 +137,9 @@ RabbitMQ.Stream.Client.PublishFilter.SizeNeeded.get -> int
 RabbitMQ.Stream.Client.PublishFilter.Write(System.Span<byte> span) -> int
 RabbitMQ.Stream.Client.RawConsumer.Info.get -> RabbitMQ.Stream.Client.ConsumerInfo
 RabbitMQ.Stream.Client.RawProducer.Info.get -> RabbitMQ.Stream.Client.ProducerInfo
-RabbitMQ.Stream.Client.RawSuperStreamConsumer.Close(bool ignoreIfClosed = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+RabbitMQ.Stream.Client.RawSuperStreamConsumer.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
 RabbitMQ.Stream.Client.RawSuperStreamConsumer.Info.get -> RabbitMQ.Stream.Client.ConsumerInfo
-RabbitMQ.Stream.Client.RawSuperStreamProducer.Close(bool ignoreIfClosed = false) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
+RabbitMQ.Stream.Client.RawSuperStreamProducer.Close() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.ResponseCode>
 RabbitMQ.Stream.Client.RawSuperStreamProducer.Info.get -> RabbitMQ.Stream.Client.ProducerInfo
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.get -> RabbitMQ.Stream.Client.RoutingStrategyType
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.set -> void