First pass cleanup

bording · bording · commit a150b025ef0b · 2025-02-28T18:06:42.000-05:00
diff --git a/src/ServiceControl.Transports.RabbitMQ/RabbitMQQuery.cs b/src/ServiceControl.Transports.RabbitMQ/RabbitMQQuery.cs
@@ -19,18 +19,15 @@ namespace ServiceControl.Transports.RabbitMQ;
 
 public class RabbitMQQuery : BrokerThroughputQuery
 {
-    readonly ResiliencePipeline pipeline = new ResiliencePipelineBuilder()
-        .AddRetry(new RetryStrategyOptions()) // Add retry using the default options
-        .AddTimeout(TimeSpan.FromMinutes(2)) // Add timeout if it keeps failing
-        .Build();
     readonly ILogger<RabbitMQQuery> logger;
     readonly TimeProvider timeProvider;
     readonly RabbitMQTransport rabbitMQTransport;
+    readonly ResiliencePipeline pipeline = new ResiliencePipelineBuilder()
+      .AddRetry(new RetryStrategyOptions()) // Add retry using the default options
+      .AddTimeout(TimeSpan.FromMinutes(2)) // Add timeout if it keeps failing
+      .Build();
 
-    public RabbitMQQuery(ILogger<RabbitMQQuery> logger,
-        TimeProvider timeProvider,
-        TransportSettings transportSettings,
-        ITransportCustomization transportCustomization) : base(logger, "RabbitMQ")
+    public RabbitMQQuery(ILogger<RabbitMQQuery> logger, TimeProvider timeProvider, TransportSettings transportSettings, ITransportCustomization transportCustomization) : base(logger, "RabbitMQ")
     {
         this.logger = logger;
         this.timeProvider = timeProvider;
@@ -65,16 +62,10 @@ void CheckLegacySettings(ReadOnlyDictionary<string, string> settings, string key
         }
     }
 
-    public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBrokerQueue brokerQueue,
-        DateOnly startDate,
-        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBrokerQueue brokerQueue, DateOnly startDate, [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         var queue = (RabbitMQBrokerQueueDetails)brokerQueue;
-        //var url = $"/api/queues/{HttpUtility.UrlEncode(queue.VHost)}/{HttpUtility.UrlEncode(queue.QueueName)}";
-
-        //logger.LogDebug($"Querying {url}");
-
-        var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, cancellationToken), cancellationToken);
+        var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, token), cancellationToken);
 
         if (response.Value is null)
         {
@@ -89,8 +80,8 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro
         for (var i = 0; i < 24 * 4; i++)
         {
             await Task.Delay(TimeSpan.FromMinutes(15), timeProvider, cancellationToken);
-            //logger.LogDebug($"Querying {url}");
-            response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, cancellationToken), cancellationToken);
+
+            response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, token), cancellationToken);
 
             if (response.Value is null)
             {
@@ -99,6 +90,7 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro
 
             newReading = new RabbitMQBrokerQueueDetails(response.Value);
             var newTotalThroughput = queue.CalculateThroughputFrom(newReading);
+
             yield return new QueueThroughput
             {
                 DateUTC = DateOnly.FromDateTime(timeProvider.GetUtcNow().DateTime),
@@ -107,12 +99,9 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro
         }
     }
 
-    async Task GetRabbitDetails(bool skipResiliencePipeline, CancellationToken cancellationToken)
+    async Task GetRabbitDetails(CancellationToken cancellationToken)
     {
-
-        var response = skipResiliencePipeline
-            ? await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken)
-            : await pipeline.ExecuteAsync(async async => await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken), cancellationToken);
+        var response = await pipeline.ExecuteAsync(async async => await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken), cancellationToken);
 
         ValidateResponse(response);
 
@@ -142,15 +131,14 @@ public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames([EnumeratorCa
     {
         var page = 1;
         bool morePages;
-        //var vHosts = new HashSet<string>(StringComparer.CurrentCultureIgnoreCase);
 
-        await GetRabbitDetails(false, cancellationToken);
+        await GetRabbitDetails(cancellationToken);
 
         do
         {
             (var queues, morePages) = await GetPage(page, cancellationToken);
 
-            if (queues != null)
+            if (queues is not null)
             {
                 foreach (var rabbitMQQueueDetails in queues)
                 {
@@ -160,27 +148,24 @@ public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames([EnumeratorCa
                     {
                         continue;
                     }
-                    //vHosts.Add(rabbitMQQueueDetails.VHost);
+
                     await AddAdditionalQueueDetails(rabbitMQQueueDetails, cancellationToken);
                     yield return rabbitMQQueueDetails;
                 }
             }
 
             page++;
         } while (morePages);
-
-        //ScopeType = vHosts.Count > 1 ? "VirtualHost" : null;
     }
 
     async Task AddAdditionalQueueDetails(RabbitMQBrokerQueueDetails brokerQueue, CancellationToken cancellationToken)
     {
         try
         {
-            var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForQueue(brokerQueue.QueueName, cancellationToken), cancellationToken);
+            var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForQueue(brokerQueue.QueueName, token), cancellationToken);
 
             // Check if conventional binding is found
             if (response.Value.Any(binding => binding?.Source == brokerQueue.QueueName
-                //&& binding?.Vhost == brokerQueue.VHost
                 && binding?.Destination == brokerQueue.QueueName
                 && binding?.DestinationType == "queue"
                 && binding?.RoutingKey == string.Empty
@@ -196,11 +181,10 @@ async Task AddAdditionalQueueDetails(RabbitMQBrokerQueueDetails brokerQueue, Can
 
         try
         {
-            var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForExchange(brokerQueue.QueueName, cancellationToken), cancellationToken);
+            var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForExchange(brokerQueue.QueueName, token), cancellationToken);
 
             // Check if delayed binding is found
             if (response.Value.Any(binding => binding?.Source is "nsb.v2.delay-delivery" or "nsb.delay-delivery"
-                    //&& binding?.Vhost == brokerQueue.VHost
                     && binding?.Destination == brokerQueue.QueueName
                     && binding?.DestinationType == "exchange"
                     && binding?.RoutingKey == $"#.{brokerQueue.QueueName}"))
@@ -216,7 +200,7 @@ async Task AddAdditionalQueueDetails(RabbitMQBrokerQueueDetails brokerQueue, Can
 
     internal async Task<(List<RabbitMQBrokerQueueDetails>?, bool morePages)> GetPage(int page, CancellationToken cancellationToken)
     {
-        var (StatusCode, Reason, Value, MorePages) = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueues(page, 500, cancellationToken), cancellationToken);
+        var (StatusCode, Reason, Value, MorePages) = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueues(page, 500, token), cancellationToken);
 
         ValidateResponse((StatusCode, Reason, Value));
 
@@ -241,19 +225,25 @@ static List<RabbitMQBrokerQueueDetails> MaterializeQueueDetails(List<Queue> item
         new KeyDescriptionPair(RabbitMQSettings.Password, RabbitMQSettings.PasswordDescription)
     ];
 
-    protected override async Task<(bool Success, List<string> Errors)> TestConnectionCore(
-        CancellationToken cancellationToken)
+    protected override async Task<(bool Success, List<string> Errors)> TestConnectionCore(CancellationToken cancellationToken)
     {
         try
         {
-            await GetRabbitDetails(true, cancellationToken);
+            var (statusCode, reason, value) = await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken);
+
+            if (value is not null)
+            {
+                return (true, []);
+            }
+            else
+            {
+                return (false, [$"{statusCode}: {reason}"]);
+            }
         }
-        catch (HttpRequestException e)
+        catch (HttpRequestException ex)
         {
-            throw new Exception($"Failed to connect to management API", e);
+            throw new Exception($"Failed to connect to RabbitMQ management API", ex);
         }
-
-        return (true, []);
     }
 
     public static class RabbitMQSettings

Original file line number	Diff line number	Diff line change
`@@ -19,18 +19,15 @@ namespace ServiceControl.Transports.RabbitMQ;`
`19`	`19`
`20`	`20`	`public class RabbitMQQuery : BrokerThroughputQuery`
`21`	`21`	`{`
`22`		`- readonly ResiliencePipeline pipeline = new ResiliencePipelineBuilder()`
`23`		`- .AddRetry(new RetryStrategyOptions()) // Add retry using the default options`
`24`		`- .AddTimeout(TimeSpan.FromMinutes(2)) // Add timeout if it keeps failing`
`25`		`- .Build();`
`26`	`22`	`readonly ILogger<RabbitMQQuery> logger;`
`27`	`23`	`readonly TimeProvider timeProvider;`
`28`	`24`	`readonly RabbitMQTransport rabbitMQTransport;`
	`25`	`+ readonly ResiliencePipeline pipeline = new ResiliencePipelineBuilder()`
	`26`	`+ .AddRetry(new RetryStrategyOptions()) // Add retry using the default options`
	`27`	`+ .AddTimeout(TimeSpan.FromMinutes(2)) // Add timeout if it keeps failing`
	`28`	`+ .Build();`
`29`	`29`
`30`		`- public RabbitMQQuery(ILogger<RabbitMQQuery> logger,`
`31`		`- TimeProvider timeProvider,`
`32`		`- TransportSettings transportSettings,`
`33`		`- ITransportCustomization transportCustomization) : base(logger, "RabbitMQ")`
	`30`	`+ public RabbitMQQuery(ILogger<RabbitMQQuery> logger, TimeProvider timeProvider, TransportSettings transportSettings, ITransportCustomization transportCustomization) : base(logger, "RabbitMQ")`
`34`	`31`	`{`
`35`	`32`	`this.logger = logger;`
`36`	`33`	`this.timeProvider = timeProvider;`
`@@ -65,16 +62,10 @@ void CheckLegacySettings(ReadOnlyDictionary<string, string> settings, string key`
`65`	`62`	`}`
`66`	`63`	`}`
`67`	`64`
`68`		`- public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBrokerQueue brokerQueue,`
`69`		`- DateOnly startDate,`
`70`		`- [EnumeratorCancellation] CancellationToken cancellationToken = default)`
	`65`	`+ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBrokerQueue brokerQueue, DateOnly startDate, [EnumeratorCancellation] CancellationToken cancellationToken = default)`
`71`	`66`	`{`
`72`	`67`	`var queue = (RabbitMQBrokerQueueDetails)brokerQueue;`
`73`		`- //var url = $"/api/queues/{HttpUtility.UrlEncode(queue.VHost)}/{HttpUtility.UrlEncode(queue.QueueName)}";`
`74`		`-`
`75`		`- //logger.LogDebug($"Querying {url}");`
`76`		`-`
`77`		`- var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, cancellationToken), cancellationToken);`
	`68`	`+ var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, token), cancellationToken);`
`78`	`69`
`79`	`70`	`if (response.Value is null)`
`80`	`71`	`{`
`@@ -89,8 +80,8 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro`
`89`	`80`	`for (var i = 0; i < 24 * 4; i++)`
`90`	`81`	`{`
`91`	`82`	`await Task.Delay(TimeSpan.FromMinutes(15), timeProvider, cancellationToken);`
`92`		`- //logger.LogDebug($"Querying {url}");`
`93`		`- response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, cancellationToken), cancellationToken);`
	`83`	`+`
	`84`	`+ response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueue(queue.QueueName, token), cancellationToken);`
`94`	`85`
`95`	`86`	`if (response.Value is null)`
`96`	`87`	`{`
`@@ -99,6 +90,7 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro`
`99`	`90`
`100`	`91`	`newReading = new RabbitMQBrokerQueueDetails(response.Value);`
`101`	`92`	`var newTotalThroughput = queue.CalculateThroughputFrom(newReading);`
	`93`	`+`
`102`	`94`	`yield return new QueueThroughput`
`103`	`95`	`{`
`104`	`96`	`DateUTC = DateOnly.FromDateTime(timeProvider.GetUtcNow().DateTime),`
`@@ -107,12 +99,9 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro`
`107`	`99`	`}`
`108`	`100`	`}`
`109`	`101`
`110`		`- async Task GetRabbitDetails(bool skipResiliencePipeline, CancellationToken cancellationToken)`
	`102`	`+ async Task GetRabbitDetails(CancellationToken cancellationToken)`
`111`	`103`	`{`
`112`		`-`
`113`		`- var response = skipResiliencePipeline`
`114`		`- ? await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken)`
`115`		`- : await pipeline.ExecuteAsync(async async => await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken), cancellationToken);`
	`104`	`+ var response = await pipeline.ExecuteAsync(async async => await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken), cancellationToken);`
`116`	`105`
`117`	`106`	`ValidateResponse(response);`
`118`	`107`
`@@ -142,15 +131,14 @@ public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames([EnumeratorCa`
`142`	`131`	`{`
`143`	`132`	`var page = 1;`
`144`	`133`	`bool morePages;`
`145`		`- //var vHosts = new HashSet<string>(StringComparer.CurrentCultureIgnoreCase);`
`146`	`134`
`147`		`- await GetRabbitDetails(false, cancellationToken);`
	`135`	`+ await GetRabbitDetails(cancellationToken);`
`148`	`136`
`149`	`137`	`do`
`150`	`138`	`{`
`151`	`139`	`(var queues, morePages) = await GetPage(page, cancellationToken);`
`152`	`140`
`153`		`- if (queues != null)`
	`141`	`+ if (queues is not null)`
`154`	`142`	`{`
`155`	`143`	`foreach (var rabbitMQQueueDetails in queues)`
`156`	`144`	`{`
`@@ -160,27 +148,24 @@ public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames([EnumeratorCa`
`160`	`148`	`{`
`161`	`149`	`continue;`
`162`	`150`	`}`
`163`		`- //vHosts.Add(rabbitMQQueueDetails.VHost);`
	`151`	`+`
`164`	`152`	`await AddAdditionalQueueDetails(rabbitMQQueueDetails, cancellationToken);`
`165`	`153`	`yield return rabbitMQQueueDetails;`
`166`	`154`	`}`
`167`	`155`	`}`
`168`	`156`
`169`	`157`	`page++;`
`170`	`158`	`} while (morePages);`
`171`		`-`
`172`		`- //ScopeType = vHosts.Count > 1 ? "VirtualHost" : null;`
`173`	`159`	`}`
`174`	`160`
`175`	`161`	`async Task AddAdditionalQueueDetails(RabbitMQBrokerQueueDetails brokerQueue, CancellationToken cancellationToken)`
`176`	`162`	`{`
`177`	`163`	`try`
`178`	`164`	`{`
`179`		`- var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForQueue(brokerQueue.QueueName, cancellationToken), cancellationToken);`
	`165`	`+ var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForQueue(brokerQueue.QueueName, token), cancellationToken);`
`180`	`166`
`181`	`167`	`// Check if conventional binding is found`
`182`	`168`	`if (response.Value.Any(binding => binding?.Source == brokerQueue.QueueName`
`183`		`- //&& binding?.Vhost == brokerQueue.VHost`
`184`	`169`	`&& binding?.Destination == brokerQueue.QueueName`
`185`	`170`	`&& binding?.DestinationType == "queue"`
`186`	`171`	`&& binding?.RoutingKey == string.Empty`
`@@ -196,11 +181,10 @@ async Task AddAdditionalQueueDetails(RabbitMQBrokerQueueDetails brokerQueue, Can`
`196`	`181`
`197`	`182`	`try`
`198`	`183`	`{`
`199`		`- var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForExchange(brokerQueue.QueueName, cancellationToken), cancellationToken);`
	`184`	`+ var response = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetBindingsForExchange(brokerQueue.QueueName, token), cancellationToken);`
`200`	`185`
`201`	`186`	`// Check if delayed binding is found`
`202`	`187`	`if (response.Value.Any(binding => binding?.Source is "nsb.v2.delay-delivery" or "nsb.delay-delivery"`
`203`		`- //&& binding?.Vhost == brokerQueue.VHost`
`204`	`188`	`&& binding?.Destination == brokerQueue.QueueName`
`205`	`189`	`&& binding?.DestinationType == "exchange"`
`206`	`190`	`&& binding?.RoutingKey == $"#.{brokerQueue.QueueName}"))`
`@@ -216,7 +200,7 @@ async Task AddAdditionalQueueDetails(RabbitMQBrokerQueueDetails brokerQueue, Can`
`216`	`200`
`217`	`201`	`internal async Task<(List<RabbitMQBrokerQueueDetails>?, bool morePages)> GetPage(int page, CancellationToken cancellationToken)`
`218`	`202`	`{`
`219`		`- var (StatusCode, Reason, Value, MorePages) = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueues(page, 500, cancellationToken), cancellationToken);`
	`203`	`+ var (StatusCode, Reason, Value, MorePages) = await pipeline.ExecuteAsync(async token => await rabbitMQTransport.ManagementClient.GetQueues(page, 500, token), cancellationToken);`
`220`	`204`
`221`	`205`	`ValidateResponse((StatusCode, Reason, Value));`
`222`	`206`
`@@ -241,19 +225,25 @@ static List<RabbitMQBrokerQueueDetails> MaterializeQueueDetails(List<Queue> item`
`241`	`225`	`new KeyDescriptionPair(RabbitMQSettings.Password, RabbitMQSettings.PasswordDescription)`
`242`	`226`	`];`
`243`	`227`
`244`		`- protected override async Task<(bool Success, List<string> Errors)> TestConnectionCore(`
`245`		`- CancellationToken cancellationToken)`
	`228`	`+ protected override async Task<(bool Success, List<string> Errors)> TestConnectionCore(CancellationToken cancellationToken)`
`246`	`229`	`{`
`247`	`230`	`try`
`248`	`231`	`{`
`249`		`- await GetRabbitDetails(true, cancellationToken);`
	`232`	`+ var (statusCode, reason, value) = await rabbitMQTransport.ManagementClient.GetOverview(cancellationToken);`
	`233`	`+`
	`234`	`+ if (value is not null)`
	`235`	`+ {`
	`236`	`+ return (true, []);`
	`237`	`+ }`
	`238`	`+ else`
	`239`	`+ {`
	`240`	`+ return (false, [$"{statusCode}: {reason}"]);`
	`241`	`+ }`
`250`	`242`	`}`
`251`		`- catch (HttpRequestException e)`
	`243`	`+ catch (HttpRequestException ex)`
`252`	`244`	`{`
`253`		`- throw new Exception($"Failed to connect to management API", e);`
	`245`	`+ throw new Exception($"Failed to connect to RabbitMQ management API", ex);`
`254`	`246`	`}`
`255`		`-`
`256`		`- return (true, []);`
`257`	`247`	`}`
`258`	`248`
`259`	`249`	`public static class RabbitMQSettings`