remove lock for events

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/Client.cs

@@ -205,26 +205,6 @@ private async Task OnConnectionClosed(string reason)
             }
         }
 
-        private readonly SemaphoreSlim _attachSemaphore = new(1, 1);
-
-        public void AttachEventsToTheClient(ConnectionCloseHandler connectionCloseHandler,
-            ClientParameters.MetadataUpdateHandler metadataUpdateHandler)
-        {
-            _attachSemaphore.Wait();
-            ConnectionClosed += connectionCloseHandler;
-            Parameters.OnMetadataUpdate += metadataUpdateHandler;
-            _attachSemaphore.Release();
-        }
-
-        public void DetachEventsFromTheClient(ConnectionCloseHandler connectionCloseHandler,
-            ClientParameters.MetadataUpdateHandler metadataUpdateHandler)
-        {
-            _attachSemaphore.Wait();
-            ConnectionClosed -= connectionCloseHandler;
-            Parameters.OnMetadataUpdate -= metadataUpdateHandler;
-            _attachSemaphore.Release();
-        }
-
         public static async Task<Client> Create(ClientParameters parameters, ILogger logger = null)
         {
             var client = new Client(parameters, logger);

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -28,13 +28,11 @@ RabbitMQ.Stream.Client.AuthMechanismNotSupportedException.AuthMechanismNotSuppor
 RabbitMQ.Stream.Client.Chunk.Crc.get -> uint
 RabbitMQ.Stream.Client.Chunk.Data.get -> System.ReadOnlyMemory<byte>
 RabbitMQ.Stream.Client.Chunk.MagicVersion.get -> byte
-RabbitMQ.Stream.Client.Client.AttachEventsToTheClient(RabbitMQ.Stream.Client.Client.ConnectionCloseHandler connectionCloseHandler, RabbitMQ.Stream.Client.ClientParameters.MetadataUpdateHandler metadataUpdateHandler) -> void
 RabbitMQ.Stream.Client.Client.ClientId.get -> string
 RabbitMQ.Stream.Client.Client.ClientId.init -> void
 RabbitMQ.Stream.Client.Client.Consumers.get -> System.Collections.Generic.IDictionary<byte, (string, RabbitMQ.Stream.Client.ConsumerEvents)>
 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.DetachEventsFromTheClient(RabbitMQ.Stream.Client.Client.ConnectionCloseHandler connectionCloseHandler, RabbitMQ.Stream.Client.ClientParameters.MetadataUpdateHandler metadataUpdateHandler) -> void
 RabbitMQ.Stream.Client.Client.ExchangeVersions() -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.CommandVersionsResponse>
 RabbitMQ.Stream.Client.Client.Publishers.get -> System.Collections.Generic.IDictionary<byte, (string, (System.Action<System.ReadOnlyMemory<ulong>>, System.Action<(ulong, RabbitMQ.Stream.Client.ResponseCode)[]>))>
 RabbitMQ.Stream.Client.Client.QueryRoute(string superStream, string routingKey) -> System.Threading.Tasks.Task<RabbitMQ.Stream.Client.RouteQueryResponse>

RabbitMQ.Stream.Client/RawConsumer.cs

@@ -587,7 +587,9 @@ internal async Task Init()
 
             if (response.ResponseCode == ResponseCode.Ok)
             {
-                _client.AttachEventsToTheClient(OnConnectionClosed(), OnMetadataUpdate());
+                _client.ConnectionClosed += OnConnectionClosed();
+                _client.Parameters.OnMetadataUpdate += OnMetadataUpdate();
+
                 _status = EntityStatus.Open;
                 // the subscription is completed so the parsechunk can start to process the chunks
                 _completeSubscription.SetResult();
@@ -609,7 +611,8 @@ private ClientParameters.MetadataUpdateHandler OnMetadataUpdate() =>
                 // remove the event since the consumer is closed
                 // only if the stream is the valid
 
-                _client.DetachEventsFromTheClient(OnConnectionClosed(), OnMetadataUpdate());
+                _client.ConnectionClosed -= OnConnectionClosed();
+                _client.Parameters.OnMetadataUpdate -= OnMetadataUpdate();
 
                 // at this point the server has removed the consumer from the list 
                 // and the unsubscribe is not needed anymore (ignoreIfClosed = true)
@@ -623,7 +626,9 @@ private ClientParameters.MetadataUpdateHandler OnMetadataUpdate() =>
         private Client.ConnectionCloseHandler OnConnectionClosed() =>
             async reason =>
             {
-                _client.DetachEventsFromTheClient(OnConnectionClosed(), OnMetadataUpdate());
+                _client.ConnectionClosed -= OnConnectionClosed();
+                _client.Parameters.OnMetadataUpdate -= OnMetadataUpdate();
+
                 // remove the event since the connection is closed
                 _config.Pool.Remove(_client.ClientId);
                 UpdateStatusToClosed();

RabbitMQ.Stream.Client/RawProducer.cs

@@ -151,7 +151,8 @@ private async Task Init()
 
             if (response.ResponseCode == ResponseCode.Ok)
             {
-                _client.AttachEventsToTheClient(OnConnectionClosed(), OnMetadataUpdate());
+                _client.ConnectionClosed += OnConnectionClosed();
+                _client.Parameters.OnMetadataUpdate += OnMetadataUpdate();
                 _status = EntityStatus.Open;
                 return;
             }
@@ -183,7 +184,8 @@ private ClientParameters.MetadataUpdateHandler OnMetadataUpdate() =>
                 if (metaDataUpdate.Stream != _config.Stream)
                     return;
 
-                _client.DetachEventsFromTheClient(OnConnectionClosed(), OnMetadataUpdate());
+                _client.ConnectionClosed -= OnConnectionClosed();
+                _client.Parameters.OnMetadataUpdate -= OnMetadataUpdate();
 
                 _config.Pool.RemoveProducerEntityFromStream(_client.ClientId, EntityId, _config.Stream);
 

RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs

@@ -57,7 +57,7 @@ private async Task<IConsumer> StandardConsumer(bool boot)
             {
                 if (closeReason == ConnectionClosedReason.Normal)
                 {
-                    BaseLogger.LogInformation("Reconnect is skipped. {Identity} is closed normally", ToString());
+                    BaseLogger.LogInformation("{Identity} is closed normally", ToString());
                     return;
                 }
 

RabbitMQ.Stream.Client/Reliable/ProducerFactory.cs

@@ -75,7 +75,7 @@ private async Task<IProducer> StandardProducer()
             {
                 if (closeReason == ConnectionClosedReason.Normal)
                 {
-                    BaseLogger.LogInformation("Reconnect is skipped. {Identity} is closed normally", ToString());
+                    BaseLogger.LogInformation("{Identity} is closed normally", ToString());
                     return;
                 }