formatting

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

Author: Gsantomaggio

RabbitMQ.Stream.Client/Client.cs

@@ -354,7 +354,7 @@ await Request<DeletePublisherRequest, DeletePublisherResponse>(corr =>
             Dictionary<string, string> properties, Func<Deliver, Task> deliverHandler,
             Func<bool, Task<IOffsetType>> consumerUpdateHandler = null)
         {
-            return await Subscribe(new RawConsumerConfig(stream) {OffsetSpec = offsetType},
+            return await Subscribe(new RawConsumerConfig(stream) { OffsetSpec = offsetType },
                 initialCredit,
                 properties,
                 deliverHandler,
@@ -713,7 +713,7 @@ public async Task<CloseResponse> Close(string reason)
         // if the client has entities (publishers or consumers) it will be released from the pool
         // Release will decrement the active ids for the connection
         // if the active ids are 0 the connection will be closed
-        internal async Task<CloseResponse> MaybeClose(string reason,string stream, ConnectionsPool pool)
+        internal async Task<CloseResponse> MaybeClose(string reason, string stream, ConnectionsPool pool)
         {
             if (!string.IsNullOrEmpty(ClientId))
             {
@@ -758,9 +758,9 @@ public async ValueTask<MetaDataResponse> QueryMetadata(string[] streams)
 
         public async Task<bool> StreamExists(string stream)
         {
-            var streams = new[] {stream};
+            var streams = new[] { stream };
             var response = await QueryMetadata(streams).ConfigureAwait(false);
-            return response.StreamInfos is {Count: >= 1} &&
+            return response.StreamInfos is { Count: >= 1 } &&
                    response.StreamInfos[stream].ResponseCode == ResponseCode.Ok;
         }
 
@@ -807,7 +807,7 @@ public static ManualResetValueTaskSource<T> Rent()
             }
             else
             {
-                return new ManualResetValueTaskSource<T>() {RunContinuationsAsynchronously = true};
+                return new ManualResetValueTaskSource<T>() { RunContinuationsAsynchronously = true };
             }
         }
 

RabbitMQ.Stream.Client/ConnectionsPool.cs

@@ -232,10 +232,10 @@ public void Remove(string clientId)
     public int ConnectionsCount => Connections.Count;
 
     public int ActiveIdsCount => Connections.Values.Sum(x => x.StreamIds.Values.Sum(y => y.Count));
-    
+
     public int ActiveIdsCountForStream(string stream) => Connections.Values.Sum(x => x.StreamIds.TryGetValue(stream, out var streamIds) ? streamIds.Count : 0);
-    
+
     public int ActiveIdsCountForClient(string clientId) => Connections.TryGetValue(clientId, out var connectionItem) ? connectionItem.StreamIds.Values.Sum(y => y.Count) : 0;
-    
+
     public int ActiveIdsCountForClientAndStream(string clientId, string stream) => Connections.TryGetValue(clientId, out var connectionItem) && connectionItem.StreamIds.TryGetValue(stream, out var streamIds) ? streamIds.Count : 0;
 }

RabbitMQ.Stream.Client/RawConsumer.cs

@@ -87,14 +87,14 @@ internal void Validate()
 
             switch (ConsumerFilter)
             {
-                case {PostFilter: null}:
+                case { PostFilter: null }:
                     throw new ArgumentException("PostFilter must be provided when Filter is set");
-                case {Values.Count: 0}:
+                case { Values.Count: 0 }:
                     throw new ArgumentException("Values must be provided when Filter is set");
             }
         }
 
-        internal bool IsFiltering => ConsumerFilter is {Values.Count: > 0};
+        internal bool IsFiltering => ConsumerFilter is { Values.Count: > 0 };
 
         // it is needed to be able to add the subscriptions arguments
         // see consumerProperties["super-stream"] = SuperStream;

RabbitMQ.Stream.Client/RawProducer.cs

@@ -42,7 +42,7 @@ public RawProducerConfig(string stream)
 
         internal void Validate()
         {
-            if (Filter is {FilterValue: not null} && !AvailableFeaturesSingleton.Instance.PublishFilter)
+            if (Filter is { FilterValue: not null } && !AvailableFeaturesSingleton.Instance.PublishFilter)
             {
                 throw new UnsupportedOperationException(Consts.FilterNotSupported);
             }
@@ -128,7 +128,9 @@ private async Task Init()
                         {
                             _config.ConfirmHandler(new Confirmation
                             {
-                                PublishingId = id, Code = ResponseCode.Ok, Stream = _config.Stream
+                                PublishingId = id,
+                                Code = ResponseCode.Ok,
+                                Stream = _config.Stream
                             });
                         }
                         catch (Exception e)
@@ -148,7 +150,7 @@ private async Task Init()
                 {
                     foreach (var (id, code) in errors)
                     {
-                        _config.ConfirmHandler(new Confirmation {PublishingId = id, Code = code,});
+                        _config.ConfirmHandler(new Confirmation { PublishingId = id, Code = code, });
                     }
 
                     _semaphore.Release(errors.Length);
@@ -164,7 +166,7 @@ private async Task Init()
             throw new CreateProducerException($"producer could not be created code: {response.ResponseCode}");
         }
 
-        private bool IsFilteringEnabled => _config.Filter is {FilterValue: not null};
+        private bool IsFilteringEnabled => _config.Filter is { FilterValue: not null };
 
         /// <summary>
         /// SubEntry Batch send: Aggregate more messages under the same publishingId.
@@ -362,7 +364,7 @@ public async Task<ResponseCode> Close()
                 _logger.LogError(e, "Error removing the producer id: {PublisherId} from the server", _publisherId);
             }
 
-            var closed = await _client.MaybeClose($"client-close-publisher: {_publisherId}", 
+            var closed = await _client.MaybeClose($"client-close-publisher: {_publisherId}",
                     _config.Stream, _config.Pool)
                 .ConfigureAwait(false);
             ClientExceptions.MaybeThrowException(closed.ResponseCode, $"client-close-publisher: {_publisherId}");

RabbitMQ.Stream.Client/RoutingClient.cs

@@ -67,15 +67,15 @@ internal static async Task<IClient> LookupConnection(
                 // In this case we just return the node (leader for producer, random for consumer)
                 // since there is not load balancer configuration
 
-                return await routing.CreateClient(clientParameters with {Endpoint = endPointNoLb}, broker, logger)
+                return await routing.CreateClient(clientParameters with { Endpoint = endPointNoLb }, broker, logger)
                     .ConfigureAwait(false);
             }
 
             // here it means that there is a AddressResolver configuration
             // so there is a load-balancer or proxy we need to get the right connection
             // as first we try with the first node given from the LB
             var endPoint = clientParameters.AddressResolver.EndPoint;
-            var client = await routing.CreateClient(clientParameters with {Endpoint = endPoint}, broker, logger)
+            var client = await routing.CreateClient(clientParameters with { Endpoint = endPoint }, broker, logger)
                 .ConfigureAwait(false);
 
             var advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
@@ -87,7 +87,7 @@ internal static async Task<IClient> LookupConnection(
                 attemptNo++;
                 await client.Close("advertised_host or advertised_port doesn't match").ConfigureAwait(false);
 
-                client = await routing.CreateClient(clientParameters with {Endpoint = endPoint}, broker, logger)
+                client = await routing.CreateClient(clientParameters with { Endpoint = endPoint }, broker, logger)
                     .ConfigureAwait(false);
 
                 advertisedHost = GetPropertyValue(client.ConnectionProperties, "advertised_host");
@@ -156,7 +156,7 @@ private static string GetPropertyValue(IDictionary<string, string> connectionPro
         public static async Task<IClient> LookupLeaderConnection(ClientParameters clientParameters,
             StreamInfo metaDataInfo, ConnectionsPool pool, ILogger logger = null)
         {
-            return await pool.GetOrCreateClient(metaDataInfo.Leader.ToString(),metaDataInfo.Stream,
+            return await pool.GetOrCreateClient(metaDataInfo.Leader.ToString(), metaDataInfo.Stream,
                 async () =>
                     await LookupConnection(clientParameters, metaDataInfo.Leader, MaxAttempts(metaDataInfo), logger)
                         .ConfigureAwait(false)).ConfigureAwait(false);
@@ -168,7 +168,7 @@ await LookupConnection(clientParameters, metaDataInfo.Leader, MaxAttempts(metaDa
         public static async Task<IClient> LookupRandomConnection(ClientParameters clientParameters,
             StreamInfo metaDataInfo, ConnectionsPool pool, ILogger logger = null)
         {
-            var brokers = new List<Broker>() {metaDataInfo.Leader};
+            var brokers = new List<Broker>() { metaDataInfo.Leader };
             brokers.AddRange(metaDataInfo.Replicas);
             brokers.Sort((_, _) => Random.Shared.Next(-1, 1));
             var exceptions = new List<Exception>();