Add command line parsing and rabbitmq classic queue support (#338)

* Add command line parsing and quorum queue support

* Acceptance tests

* Add queueType parameter to RabbitMQ transport configuration in acceptance tests

* Fix misleading assert.ignore

* Refactor transport configuration to support asynchronous cleanup delegation in acceptance tests

* Simplify pattern

Co-authored-by: Sean Feldman <SeanFeldman@users.noreply.github.com>

* Also cleanup consumer queues in acceptance tests

* Null value handling

---------

Co-authored-by: Daniel Marbach <danielmarbach@users.noreply.github.com>
Co-authored-by: Sean Feldman <SeanFeldman@users.noreply.github.com>

Author: danielmarbach

src/ServiceControl.Connector.MassTransit.AcceptanceTesting/IConfigureTransportTestExecution.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using MassTransit;
@@ -11,7 +12,7 @@ public interface IConfigureTransportTestExecution
 {
     Func<CancellationToken, Task> ConfigureTransportForEndpoint(EndpointConfiguration endpointConfiguration, PublisherMetadata publisherMetadata);
 
-    void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator);
+    Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator);
 
-    void ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration);
+    Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration);
 }

src/ServiceControl.Connector.MassTransit.AcceptanceTesting/TestAttributes/RabbitMQTestAttribute.cs

@@ -11,7 +11,7 @@ public void ApplyToContext(TestExecutionContext context)
         var connectionString = Environment.GetEnvironmentVariable("RabbitMQTransport_ConnectionString");
         if (string.IsNullOrWhiteSpace(connectionString))
         {
-            Assert.Ignore("Ignoring because environment variable RabbitMQConnectionString is not available");
+            Assert.Ignore("Ignoring because environment variable RabbitMQTransport_ConnectionString is not available");
         }
     }
 }

src/ServiceControl.Connector.MassTransit.AcceptanceTests.AmazonSQS/ConfigureAmazonSQSTransportTestExecution.cs

@@ -14,7 +14,7 @@ public Func<CancellationToken, Task> ConfigureTransportForEndpoint(EndpointConfi
         return _ => Task.CompletedTask;
     }
 
-    public void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
     {
         var region = FallbackRegionFactory.GetRegionEndpoint().SystemName;
 
@@ -29,15 +29,17 @@ public void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurato
 
             cfg.ConfigureEndpoints(context, new DefaultEndpointNameFormatter(NamePrefixGenerator.GetNamePrefix(), false));
         });
+        return (_, _) => Task.CompletedTask;
     }
 
-    public void ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
     {
         services.UsingAmazonSqs(transport =>
         {
             transport.QueueNamePrefix = NamePrefixGenerator.GetNamePrefix();
             transport.TopicNamePrefix = NamePrefixGenerator.GetNamePrefix();
             transport.QueueNameGenerator = TestNameHelper.GetSqsQueueName;
         });
+        return (_, _) => Task.CompletedTask;
     }
 }

src/ServiceControl.Connector.MassTransit.AcceptanceTests.AzureServiceBus/ConfigureAzureServiceBusTransportTestExecution.cs

@@ -14,18 +14,20 @@ public Func<CancellationToken, Task> ConfigureTransportForEndpoint(EndpointConfi
         return Cleanup;
     }
 
-    public void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
     {
         configurator.UsingAzureServiceBus((context, cfg) =>
         {
             cfg.Host(connectionString);
             cfg.ConfigureEndpoints(context);
         });
+        return (_, _) => Task.CompletedTask;
     }
 
-    public void ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
     {
         services.UsingAzureServiceBus(configuration, connectionString);
+        return (_, _) => Task.CompletedTask;
     }
 
     Task Cleanup(CancellationToken cancellationToken)

src/ServiceControl.Connector.MassTransit.AcceptanceTests.AzureServiceBusDeadLetter/ConfigureAzureServiceBusTransportTestExecution.cs

@@ -14,7 +14,7 @@ public Func<CancellationToken, Task> ConfigureTransportForEndpoint(EndpointConfi
         return Cleanup;
     }
 
-    public void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
     {
         configurator.UsingAzureServiceBus((context, cfg) =>
         {
@@ -29,11 +29,13 @@ public void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurato
                 sb.ConfigureDeadLetterQueueErrorTransport();
             }
         });
+        return (_, _) => Task.CompletedTask;
     }
 
-    public void ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
     {
         services.UsingAzureServiceBus(configuration, connectionString, true);
+        return (_, _) => Task.CompletedTask;
     }
 
     Task Cleanup(CancellationToken cancellationToken)

src/ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ.Classic/.editorconfig

@@ -0,0 +1,9 @@
+[*.cs]
+
+# Justification: Test project
+dotnet_diagnostic.CA2007.severity = none
+dotnet_diagnostic.PS0004.severity = none  # A parameter of type CancellationToken on a private delegate or method should be required
+dotnet_diagnostic.PS0018.severity = none  # Add a CancellationToken parameter
+
+# Justification: Tests don't support cancellation and don't need to forward IMessageHandlerContext.CancellationToken
+dotnet_diagnostic.NSB0002.severity = suggestion

src/ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ.Classic/ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ.Classic.csproj

@@ -0,0 +1,32 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\ServiceControl.Connector.MassTransit.AcceptanceTesting\ServiceControl.Connector.MassTransit.AcceptanceTesting.csproj" />
+    <ProjectReference Include="..\ServiceControl.Connector.MassTransit.RabbitMQ\ServiceControl.Connector.MassTransit.RabbitMQ.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="GitHubActionsTestLogger" Version="3.0.1" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.3.0" />
+    <PackageReference Include="NUnit" Version="4.5.1" />
+    <PackageReference Include="NUnit.Analyzers" Version="4.12.0" />
+    <PackageReference Include="NUnit3TestAdapter" Version="6.1.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="MassTransit.RabbitMQ" Version="8.3.1" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <Compile Include="..\ServiceControl.Connector.MassTransit.AcceptanceTests\Shared\**\*.cs" LinkBase="Shared" />
+    <Compile Include="..\ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ\**\*.cs" Exclude="..\ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ\obj\**\*"/>
+    <Compile Remove="..\ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ\TestSuiteConfiguration.cs" />
+  </ItemGroup>
+
+</Project>

src/ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ.Classic/TestSuiteConfiguration.cs

@@ -0,0 +1,7 @@
+[assembly: RabbitMQTest]
+
+public partial class TestSuiteConfiguration
+{
+    public IConfigureTransportTestExecution CreateTransportConfiguration() => new ConfigureRabbitMQTransportTestExecution(QueueType.Classic);
+    public Task Cleanup() => Task.CompletedTask;
+}

src/ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ/ConfigureRabbitMQTransportTestExecution.cs

@@ -2,18 +2,21 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using NServiceBus.AcceptanceTesting.Support;
+using NServiceBus.Transport;
+using ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ;
 
-class ConfigureRabbitMQTransportTestExecution : IConfigureTransportTestExecution
+class ConfigureRabbitMQTransportTestExecution(QueueType queueType = QueueType.Quorum) : IConfigureTransportTestExecution
 {
+    TestRabbitMQTransport? transport;
+
     public Func<CancellationToken, Task> ConfigureTransportForEndpoint(EndpointConfiguration endpointConfiguration, PublisherMetadata publisherMetadata)
     {
-        var transport = new RabbitMQTransport(
-            RoutingTopology.Conventional(QueueType.Quorum), "host=localhost", false);
+        transport = new TestRabbitMQTransport(RoutingTopology.Conventional(queueType), "host=localhost", false);
         endpointConfiguration.UseTransport(transport);
         return Cleanup;
     }
 
-    public void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurator configurator)
     {
         configurator.UsingRabbitMq((context, cfg) =>
         {
@@ -32,19 +35,72 @@ public void ConfigureTransportForMassTransitEndpoint(IBusRegistrationConfigurato
         {
             if (cfg is IRabbitMqReceiveEndpointConfigurator rmq)
             {
-                rmq.SetQuorumQueue();
+                if (queueType == QueueType.Quorum)
+                {
+                    rmq.SetQuorumQueue();
+                }
             }
         });
+
+        return (queuesToDelete, _) =>
+        {
+            DeleteQueues(queuesToDelete);
+            return Task.CompletedTask;
+        };
     }
 
-    public void ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
+    public Func<IReadOnlyCollection<string>, CancellationToken, Task> ConfigureTransportForConnector(IServiceCollection services, IConfiguration configuration)
     {
-        services.UsingRabbitMQ("host=localhost", new Uri("http://localhost:15672/"), "guest", "guest");
+        services.UsingRabbitMQ("host=localhost", new Uri("http://localhost:15672/"), "guest", "guest", queueType);
+        return (queuesToDelete, _) =>
+        {
+            DeleteQueues(queuesToDelete);
+            return Task.CompletedTask;
+        };
     }
 
     Task Cleanup(CancellationToken cancellationToken)
     {
-        //TODO?
+        PurgeQueues();
         return Task.CompletedTask;
     }
+
+    void PurgeQueues()
+    {
+        if (transport == null)
+        {
+            return;
+        }
+
+        DeleteQueues(transport.QueuesToCleanup.ToHashSet());
+    }
+
+    static void DeleteQueues(IReadOnlyCollection<string> queues)
+    {
+        using var connection = ConnectionHelper.ConnectionFactory.CreateConnection("Test Queue Purger");
+        using var channel = connection.CreateModel();
+        foreach (var queue in queues)
+        {
+            try
+            {
+                channel.QueueDelete(queue, false, false);
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine("Unable to clear queue {0}: {1}", queue, ex);
+            }
+        }
+    }
+
+    class TestRabbitMQTransport(RoutingTopology routingTopology, string connectionString, bool enableDelayedDelivery) : RabbitMQTransport(routingTopology, connectionString, enableDelayedDelivery)
+    {
+        public override async Task<TransportInfrastructure> Initialize(HostSettings hostSettings, ReceiveSettings[] receivers, string[] sendingAddresses, CancellationToken cancellationToken = default)
+        {
+            var infrastructure = await base.Initialize(hostSettings, receivers, sendingAddresses, cancellationToken);
+            QueuesToCleanup.AddRange(infrastructure.Receivers.Select(x => x.Value.ReceiveAddress).Concat(sendingAddresses).Distinct());
+            return infrastructure;
+        }
+
+        public List<string> QueuesToCleanup { get; } = [];
+    }
 }

src/ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ/ConnectionHelper.cs

@@ -0,0 +1,32 @@
+namespace ServiceControl.Connector.MassTransit.AcceptanceTests.RabbitMQ;
+
+using System;
+using System.Security.Authentication;
+using global::RabbitMQ.Client;
+
+public static class ConnectionHelper
+{
+    static Lazy<ConnectionFactory> connectionFactory = new(() =>
+    {
+        var connectionConfiguration = AdapterRabbitMqConfiguration.ConnectionConfiguration.Create("host=localhost", "AcceptanceTests");
+
+        var factory = new ConnectionFactory
+        {
+            AutomaticRecoveryEnabled = true,
+            HostName = connectionConfiguration.Host,
+            Port = connectionConfiguration.Port,
+            VirtualHost = connectionConfiguration.VirtualHost,
+            UserName = connectionConfiguration.UserName ?? "guest",
+            Password = connectionConfiguration.Password ?? "guest"
+        };
+
+        factory.Ssl.ServerName = factory.HostName;
+        factory.Ssl.Certs = null;
+        factory.Ssl.Version = SslProtocols.Tls12;
+        factory.Ssl.Enabled = connectionConfiguration.UseTls;
+
+        return factory;
+    });
+
+    public static ConnectionFactory ConnectionFactory => connectionFactory.Value;
+}