Add Azure ServiceBus Emulator

src/AzureServiceBus.Tests/AzureServiceBus.Tests.csproj

@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <Import Project="$(CCTestProjectProps)" Condition="Exists('$(CCTestProjectProps)')"/>
+  <PropertyGroup>
+    <RootNamespace>Squadron.AzureServiceBus.Tests</RootNamespace>
+    <AssemblyName>Squadron.AzureServiceBus.Tests</AssemblyName>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\AzureServiceBus\AzureServiceBus.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <None Update="xunit.runner.json">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+  </ItemGroup>
+
+</Project>

src/AzureServiceBus.Tests/AzureServiceBusCustomConfigTests.cs

@@ -0,0 +1,45 @@
+using System;
+using System.Threading.Tasks;
+using Azure.Messaging.ServiceBus;
+using FluentAssertions;
+using Xunit;
+
+namespace Squadron.AzureServiceBus.Tests;
+
+public class AzureServiceBusCustomConfigTests(
+    AzureServiceBusResources<CustomConfig> azureServiceBusResource) :
+    IClassFixture<AzureServiceBusResources<CustomConfig>>
+{
+    [Fact]
+    public async Task Send_And_Receive()
+    {
+        var sender = azureServiceBusResource.Client
+            .CreateSender("custom.topic");
+        var serviceBusMessage = new ServiceBusMessage(BinaryData.FromString("custom_message"));
+        await sender.SendMessageAsync(serviceBusMessage);
+
+        var receiver = azureServiceBusResource.Client
+            .CreateReceiver("custom.queue");
+        var message = await receiver.ReceiveMessageAsync();
+        var receivedMessage = message.Body.ToString();
+
+        receivedMessage.Should().Be("custom_message");
+    }
+}
+
+public class CustomConfig : AzureServiceBusConfig
+{
+    protected override Queue[] CreateQueues()
+    {
+        return [new Queue("custom.queue")];
+    }
+
+    protected override Topic[] CreateTopics()
+    {
+        var subscription = new Subscription(
+            "custom.subscription", 
+            new SubscriptionProperties(ForwardTo: "custom.queue"));
+
+        return [new Topic("custom.topic", Subscriptions: [subscription])];
+    }
+}

src/AzureServiceBus.Tests/AzureServiceBusResourceTests.cs

@@ -0,0 +1,21 @@
+using System.Threading.Tasks;
+using FluentAssertions;
+using Xunit;
+
+namespace Squadron.AzureServiceBus.Tests;
+
+public class AzureServiceBusResourceTests(
+    AzureServiceBusResources azureServiceBusResource) :
+    IClassFixture<AzureServiceBusResources>
+{
+    [Fact]
+    public async Task Send_And_Receive()
+    {
+        var receiver = azureServiceBusResource.Client
+            .CreateReceiver(AzureServiceBusStatus.QueueName);
+        var message = await receiver.ReceiveMessageAsync();
+        var receivedMessage = message.Body.ToString();
+
+        receivedMessage.Should().Be("status_check");
+    }
+}

src/AzureServiceBus.Tests/xunit.runner.json

@@ -0,0 +1,4 @@
+{
+  "appDomain": "denied",
+  "parallelizeAssembly": true
+}

src/AzureServiceBus/AzureServiceBus.csproj

@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+  <Import Project="$(CCResourceProjectProps)" Condition="Exists('$(CCResourceProjectProps)')" />
+
+  <PropertyGroup>
+    <AssemblyName>Squadron.AzureServiceBus</AssemblyName>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Azure.Messaging.ServiceBus" Version="7.18.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Compose\Compose.csproj" />
+    <ProjectReference Include="..\Core\Core.csproj" />
+    <ProjectReference Include="..\SqlServer\SqlServer.csproj" />
+  </ItemGroup>
+
+</Project>

src/AzureServiceBus/AzureServiceBusConfig.cs

@@ -0,0 +1,102 @@
+using System.IO;
+using System.Linq;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace Squadron;
+
+public class AzureServiceBusConfig
+{
+    private static readonly JsonSerializerOptions _options = new()
+    {
+        DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
+    };
+
+    protected virtual Queue[] CreateQueues()
+    {
+        return [];
+    }
+
+    protected virtual Topic[] CreateTopics()
+    {
+        return [];
+    }
+
+    public string Build()
+    {
+        var fileName = Path.GetTempFileName();
+
+        Queue[] queues = CreateQueues()
+            .Concat([new Queue(AzureServiceBusStatus.QueueName)])
+            .ToArray();
+
+        var config = new
+        {
+            UserConfig = new UserConfig(
+                [new Namespace(queues, CreateTopics())],
+                new Logging())
+        };
+
+        var serializedConfig = JsonSerializer.Serialize(config, options: _options);
+        File.WriteAllText(fileName, serializedConfig);
+
+        return fileName;
+    }
+}
+
+public record UserConfig(
+    Namespace[] Namespaces,
+    Logging Logging);
+public record Namespace(
+    Queue[] Queues,
+    Topic[] Topics,
+    string Name = "sbemulatorns");
+public record Queue(
+    string Name,
+    QueueProperties? Properties = null);
+public record QueueProperties(
+    bool DeadLetteringOnMessageExpiration = true,
+    string DefaultMessageTimeToLive= "PT1H",
+    string DuplicateDetectionHistoryTimeWindow = "PT1M",
+    string ForwardDeadLetteredMessagesTo = "",
+    string ForwardTo = "",
+    string LockDuration= "PT1M",
+    int MaxDeliveryCount = 5,
+    bool RequiresDuplicateDetection = false,
+    bool RequiresSession = false);
+public record Topic(
+    string Name,
+    TopicProperties? Properties = null,
+    Subscription[]? Subscriptions = null);
+public record TopicProperties(
+    string DefaultMessageTimeToLive = "PT1H",
+    string DuplicateDetectionHistoryTimeWindow = "PT1M",
+    bool RequiresDuplicateDetection = false);
+public record Subscription(
+    string Name,
+    SubscriptionProperties? Properties = null,
+    SubscriptionRule[]? Rules = null);
+public record SubscriptionProperties(
+    bool DeadLetteringOnMessageExpiration = true,
+    string DefaultMessageTimeToLive = "PT1H",
+    string LockDuration = "PT1M",
+    int MaxDeliveryCount = 5,
+    string ForwardDeadLetteredMessagesTo = "",
+    string ForwardTo = "",
+    bool RequiresSession = false);
+public record SubscriptionRule(
+    string Name,
+    TopicRuleProperties? Properties = null);
+public record TopicRuleProperties(
+    string FilterType,
+    CorrelationFilter CorrelationFilter);
+public record CorrelationFilter(
+    string ContentType,
+    string CorrelationId,
+    string Label,
+    string MessageId,
+    string ReplyTo,
+    string ReplyToSessionId,
+    string SessionId,
+    string To);
+public record Logging(string Type = "Console");

src/AzureServiceBus/AzureServiceBusConstants.cs

@@ -0,0 +1,7 @@
+namespace Squadron;
+
+internal class AzureServiceBusConstants
+{
+    internal static string SqlServerResourceName { get; } = "asb_sql_server";
+    internal static string AzureServiceBusEmulatorResourceName { get; } = "asb_emulator";
+}

src/AzureServiceBus/AzureServiceBusDefaultOptions.cs

@@ -0,0 +1,18 @@
+namespace Squadron;
+
+public class AzureServiceBusDefaultOptions<TConfig> : ComposeResourceOptions
+    where TConfig : AzureServiceBusConfig, new()
+{
+    public override void Configure(ComposeResourceBuilder builder)
+    {
+        builder.AddContainer<SqlServerDefaultOptions>(
+            AzureServiceBusConstants.SqlServerResourceName);
+
+        builder
+            .AddContainer<AzureServiceBusEmulatorDefaultOptions<TConfig>>(
+                AzureServiceBusConstants.AzureServiceBusEmulatorResourceName)
+            .AddLink(AzureServiceBusConstants.SqlServerResourceName,
+                new EnvironmentVariableMapping("SQL_SERVER", "#NAME#"),
+                new EnvironmentVariableMapping("MSSQL_SA_PASSWORD", "#MSSQL_SA_PASSWORD#"));
+    }
+}

src/AzureServiceBus/AzureServiceBusEmulatorDefaultOptions.cs

@@ -0,0 +1,30 @@
+using System;
+
+namespace Squadron;
+
+public class AzureServiceBusEmulatorDefaultOptions<TConfig> : 
+    ContainerResourceOptions,
+    IComposableResourceOption
+    where TConfig : AzureServiceBusConfig, new()
+{
+    public Type ResourceType => typeof(AzureServiceBusEmulatorResource<TConfig>);
+
+    /// <summary>
+    /// Configure resource options for AzureServiceBusEmulatorDefaultOptions
+    /// </summary>
+    /// <param name="builder"></param>
+    public override void Configure(ContainerResourceBuilder builder)
+    {
+        var azureServiceBusConfig = new TConfig();
+        var configFile = azureServiceBusConfig.Build();
+
+        builder
+            .Name("asb_emulator")
+            .Image("mcr.microsoft.com/azure-messaging/servicebus-emulator")
+            .InternalPort(5672)
+            .WaitTimeout(120)
+            .AddEnvironmentVariable("ACCEPT_EULA=Y")
+            .AddVolume($"{configFile}:/ServiceBus_Emulator/ConfigFiles/Config.json")
+            .PreferLocalImage();
+    }
+}

src/AzureServiceBus/AzureServiceBusResource.cs

@@ -0,0 +1,46 @@
+using System;
+using System.Threading.Tasks;
+using Azure.Messaging.ServiceBus;
+using Azure.Messaging.ServiceBus.Administration;
+using Xunit;
+
+namespace Squadron;
+
+/// <inheritdoc/>
+public class AzureServiceBusEmulatorResource<TConfig>
+    : AzureServiceBusEmulatorResource<AzureServiceBusEmulatorDefaultOptions<TConfig>, TConfig> 
+    where TConfig : AzureServiceBusConfig, new();
+
+/// <summary>
+/// Represents an AzureServiceBus resource that can be used by unit tests.
+/// </summary>
+/// <seealso cref="IDisposable"/>
+public class AzureServiceBusEmulatorResource<TOptions, TConfig> : 
+    ContainerResource<TOptions>,
+    IAsyncLifetime,
+    IComposableResource
+    where TOptions : ContainerResourceOptions, new()
+{
+    /// <summary>
+    /// Connection string to access the Azure Service Bus
+    /// </summary>
+    public string ConnectionString { get; private set; }
+
+    /// <summary>
+    /// Azure Service Bus client
+    /// </summary>
+    public ServiceBusClient Client { get; private set; }
+
+    /// <inheritdoc cref="IAsyncLifetime"/>
+    public override async Task InitializeAsync()
+    {
+        await base.InitializeAsync();
+
+        ConnectionString =
+            $"Endpoint=sb://{Manager.Instance.Address}:{Manager.Instance.HostPort}" +
+            ";SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;";
+
+        Client = new ServiceBusClient(ConnectionString);
+        await Initializer.WaitAsync(new AzureServiceBusStatus(Client));
+    }
+}

src/AzureServiceBus/AzureServiceBusResources.cs

@@ -0,0 +1,25 @@
+using Azure.Messaging.ServiceBus;
+using SharpCompress;
+
+namespace Squadron;
+
+public class AzureServiceBusResources : 
+    AzureServiceBusResources<AzureServiceBusConfig>;
+
+public class AzureServiceBusResources<TConfig> : 
+    ComposeResource<AzureServiceBusDefaultOptions<TConfig>> 
+    where TConfig : AzureServiceBusConfig, new()
+{
+    private readonly Lazy<AzureServiceBusEmulatorResource<TConfig>> _azureServiceBusEmulatorResource;
+
+    public AzureServiceBusResources()
+    {
+        _azureServiceBusEmulatorResource = new Lazy<AzureServiceBusEmulatorResource<TConfig>>(() =>
+            (Managers[AzureServiceBusConstants.AzureServiceBusEmulatorResourceName].Resource
+                as AzureServiceBusEmulatorResource<TConfig>)!);
+    }
+
+    public string ConnectionString => _azureServiceBusEmulatorResource.Value.ConnectionString!;
+
+    public ServiceBusClient Client => _azureServiceBusEmulatorResource.Value.Client!;
+}

src/AzureServiceBus/AzureServiceBusStatus.cs

@@ -0,0 +1,28 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.Messaging.ServiceBus;
+
+namespace Squadron;
+
+public class AzureServiceBusStatus(ServiceBusClient client) :
+    IResourceStatusProvider
+{
+    public static string QueueName = "squadron.status.check";
+
+    public async Task<Status> IsReadyAsync(CancellationToken cancellationToken)
+    {
+        try
+        {
+            var sender = client.CreateSender(QueueName);
+            var serviceBusMessage = new ServiceBusMessage(BinaryData.FromString("status_check"));
+            await sender.SendMessageAsync(serviceBusMessage, cancellationToken);
+
+            return new Status { IsReady = true };
+        }
+        catch (Exception ex)
+        {
+            return new Status { IsReady = false, Message = "Not ready" };
+        }
+    }
+}

src/Compose.Tests/NetworkTests.cs

@@ -32,14 +32,14 @@ public async Task TwoContainer_Network_BothInSameNetwork()
             string connectionString = mongoResource.GetComposeExports()["CONNECTIONSTRING_INTERNAL"];
 
             string containerName = GetNameFromConnectionString(connectionString);
-            IList<ContainerListResponse> response = (await _dockerClient.Containers.ListContainersAsync(
-                new ContainersListParameters()));
+            IList<ContainerListResponse> response = await _dockerClient.Containers.ListContainersAsync(
+                new ContainersListParameters());
 
-            ContainerListResponse container = response.Where(c => c.Names.Contains($"/{containerName}")).Single();
+            ContainerListResponse container = response.FirstOrDefault(c => c.Names.Contains($"/{containerName}"));
 
-            string networkName = container.NetworkSettings.Networks.Keys.Where(n => n.Contains("squa_network")).Single();
+            string networkName = container?.NetworkSettings.Networks.Keys.FirstOrDefault(n => n.Contains("squa_network"));
 
-            NetworkResponse network = (await _dockerClient.Networks.ListNetworksAsync()).Where(n => n.Name == networkName).SingleOrDefault();
+            NetworkResponse network = (await _dockerClient.Networks.ListNetworksAsync()).FirstOrDefault(n => n.Name == networkName);
             network.Should().NotBeNull();
         }