Introduce AzureServiceBusHealthCheck

src/extensions/azureServiceBus/Wemogy.CQRS.Extensions.AzureServiceBus.UnitTests/Health/AzureServiceBusHealthCheckTests.cs

@@ -0,0 +1,53 @@
+using System.Threading;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+using Wemogy.Configuration;
+using Wemogy.CQRS.Extensions.AzureServiceBus.Health;
+using Xunit;
+
+namespace Wemogy.CQRS.Extensions.AzureServiceBus.UnitTests.Health;
+
+[Collection("AzureServiceBus")]
+public class AzureServiceBusHealthCheckTests
+{
+    private const string QueueName = "unit-testing-queue-1";
+    private readonly string _azureServiceBusConnectionString;
+
+    public AzureServiceBusHealthCheckTests()
+    {
+        var configuration = ConfigurationFactory.BuildConfiguration("Development");
+        _azureServiceBusConnectionString = configuration["AzureServiceBusConnectionString"] !;
+    }
+
+    [Fact]
+    public async Task CheckHealthAsync_ShouldReturnHealthyIfAzureServiceBusIsAlive()
+    {
+        // Arrange
+        var healthCheckService = new AzureServiceBusHealthCheck(_azureServiceBusConnectionString, QueueName);
+
+        // Act
+        var healthCheckResult = await healthCheckService.CheckHealthAsync(
+            new HealthCheckContext());
+
+        // Assert
+        healthCheckResult.Status.Should().Be(HealthStatus.Healthy);
+    }
+
+    [Fact]
+    public async Task CheckHealthAsync_ShouldThrowIfCancellationWasRequested()
+    {
+        // Arrange
+        var cancellationTokenSource = new CancellationTokenSource();
+        var healthCheckService = new AzureServiceBusHealthCheck(_azureServiceBusConnectionString, QueueName);
+
+        // Act
+        cancellationTokenSource.Cancel();
+        var exception = await Record.ExceptionAsync(() => healthCheckService.CheckHealthAsync(
+            new HealthCheckContext(),
+            cancellationTokenSource.Token));
+
+        // Assert
+        exception.Should().NotBeNull().And.BeOfType<TaskCanceledException>();
+    }
+}

src/extensions/azureServiceBus/Wemogy.CQRS.Extensions.AzureServiceBus/Health/AzureServiceBusHealthCheck.cs

@@ -0,0 +1,56 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Messaging.ServiceBus;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace Wemogy.CQRS.Extensions.AzureServiceBus.Health
+{
+    /// <summary>
+    /// This implementation is based on the https://github.com/Xabaril/AspNetCore.Diagnostics.HealthChecks/blob/da70571ae83a2d83f93e45c0beb2a38a633a90d6/src/HealthChecks.AzureServiceBus/AzureServiceBusQueueHealthCheck.cs
+    /// We created our own implementation, because we want to propagate the TaskCanceledException if the cancellation token has been canceled instead of returning a Unhealthy result.
+    /// </summary>
+    public class AzureServiceBusHealthCheck : IHealthCheck
+    {
+        private static readonly ConcurrentDictionary<string, ServiceBusClient> ClientConnections =
+            new ConcurrentDictionary<string, ServiceBusClient>();
+
+        private static readonly ConcurrentDictionary<string, ServiceBusReceiver> ServiceBusReceivers =
+            new ConcurrentDictionary<string, ServiceBusReceiver>();
+
+        private readonly string _queueName;
+        private readonly string _connectionString;
+
+        public AzureServiceBusHealthCheck(string connectionString, string queueName)
+        {
+            _connectionString = connectionString;
+            _queueName = queueName;
+        }
+
+        public async Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
+        {
+            try
+            {
+                var client = ClientConnections.GetOrAdd(_queueName, _ => new ServiceBusClient(_connectionString));
+                var receiver = ServiceBusReceivers.GetOrAdd($"{nameof(AzureServiceBusHealthCheck)}_{_queueName}", client.CreateReceiver(_queueName));
+                _ = await receiver.PeekMessageAsync(cancellationToken: cancellationToken);
+                return HealthCheckResult.Healthy();
+            }
+            catch (TaskCanceledException ex)
+            {
+                // propagate the exception if the cancellation token has been canceled
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    throw;
+                }
+
+                return new HealthCheckResult(context.Registration.FailureStatus, exception: ex);
+            }
+            catch (Exception ex)
+            {
+                return new HealthCheckResult(context.Registration.FailureStatus, exception: ex);
+            }
+        }
+    }
+}

src/extensions/azureServiceBus/Wemogy.CQRS.Extensions.AzureServiceBus/Health/HealthChecksBuilderExtensions.cs

@@ -0,0 +1,22 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace Wemogy.CQRS.Extensions.AzureServiceBus.Health
+{
+    public static class HealthChecksBuilderExtensions
+    {
+        public static IHealthChecksBuilder AddAzureServiceBusCheck(
+            this IHealthChecksBuilder builder,
+            string connectionString,
+            string queueName,
+            string? name = null)
+        {
+            return builder.Add(new HealthCheckRegistration(
+                name ?? nameof(AzureServiceBusHealthCheck),
+                _ => new AzureServiceBusHealthCheck(connectionString, queueName),
+                default,
+                default,
+                default));
+        }
+    }
+}