Add option to configure broker topology in DI extensions

Author: Havret

src/ArtemisNetClient.Extensions.DependencyInjection/ActiveMqArtemisClientDependencyInjectionExtensions.cs

@@ -60,7 +60,7 @@ public static IActiveMqBuilder AddActiveMq(this IServiceCollection services, str
                 var queueConfigurations = activeMqOptions.EnableQueueDeclaration ? activeMqOptions.QueueConfigurations : new List<QueueConfiguration>(0);
                 var addressConfigurations = activeMqOptions.EnableAddressDeclaration ? activeMqOptions.AddressConfigurations : new Dictionary<string, HashSet<RoutingType>>(0);
                 var lazyConnection = provider.GetConnection(name);
-                return new ActiveMqTopologyManager(lazyConnection, queueConfigurations, addressConfigurations);
+                return new ActiveMqTopologyManager(provider, lazyConnection, queueConfigurations, addressConfigurations, activeMqOptions.ConfigureTopologyActions);
             });
             builder.Services.AddSingleton(provider =>
             {
@@ -107,7 +107,7 @@ public static IActiveMqBuilder ConfigureConnectionFactory(this IActiveMqBuilder
         }
 
         /// <summary>
-        /// Adds action to configure a <see cref="IConnection"/>.
+        /// Adds an action to configure a <see cref="IConnection"/>.
         /// </summary>
         /// <param name="builder">The <see cref="IActiveMqBuilder"/>.</param>
         /// <param name="configureConnectionAction">A delegate that is used to configure a <see cref="IConnection"/>.</param>
@@ -118,6 +118,18 @@ public static IActiveMqBuilder ConfigureConnection(this IActiveMqBuilder builder
             return builder;
         }
 
+        /// <summary>
+        /// Adds an action to configure the broker topology.
+        /// </summary>
+        /// <param name="builder">The <see cref="IActiveMqBuilder"/>.</param>
+        /// <param name="configureTopologyAction">A delegate that can be used configure the broker topology.</param>
+        /// <returns>The <see cref="IActiveMqBuilder"/> that can be used to configure ActiveMQ Artemis Client.</returns>
+        public static IActiveMqBuilder ConfigureTopology(this IActiveMqBuilder builder, Func<IServiceProvider, ITopologyManager, Task> configureTopologyAction)
+        {
+            builder.Services.Configure<ActiveMqOptions>(builder.Name, options => options.ConfigureTopologyActions.Add(configureTopologyAction));
+            return builder;
+        }
+
         /// <summary>
         /// Adds the <see cref="IConsumer"/>.
         /// </summary>

src/ArtemisNetClient.Extensions.DependencyInjection/ActiveMqTopologyManager.cs

@@ -1,4 +1,5 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -8,20 +9,28 @@ namespace ActiveMQ.Artemis.Client.Extensions.DependencyInjection
 {
     internal class ActiveMqTopologyManager
     {
+        private readonly IServiceProvider _serviceProvider;
         private readonly AsyncValueLazy<IConnection> _lazyConnection;
         private readonly IReadOnlyList<QueueConfiguration> _queueConfigurations;
         private readonly IReadOnlyDictionary<string, HashSet<RoutingType>> _addressConfigurations;
+        private readonly List<Func<IServiceProvider, ITopologyManager, Task>> _configureTopologyActions;
 
-        public ActiveMqTopologyManager(AsyncValueLazy<IConnection> lazyConnection, IReadOnlyList<QueueConfiguration> queueConfigurations, IReadOnlyDictionary<string, HashSet<RoutingType>> addressConfigurations)
+        public ActiveMqTopologyManager(IServiceProvider serviceProvider,
+            AsyncValueLazy<IConnection> lazyConnection,
+            IReadOnlyList<QueueConfiguration> queueConfigurations,
+            IReadOnlyDictionary<string, HashSet<RoutingType>> addressConfigurations,
+            List<Func<IServiceProvider, ITopologyManager, Task>> configureTopologyActions)
         {
+            _serviceProvider = serviceProvider;
             _lazyConnection = lazyConnection;
             _queueConfigurations = queueConfigurations;
             _addressConfigurations = addressConfigurations;
+            _configureTopologyActions = configureTopologyActions;
         }
 
         public async Task CreateTopologyAsync(CancellationToken cancellationToken)
         {
-            if (_queueConfigurations.Count == 0 && _addressConfigurations.Count == 0)
+            if (_queueConfigurations.Count == 0 && _addressConfigurations.Count == 0 && _configureTopologyActions.Count == 0)
             {
                 return;
             }
@@ -30,6 +39,11 @@ public async Task CreateTopologyAsync(CancellationToken cancellationToken)
             var topologyManager = await connection.CreateTopologyManagerAsync(cancellationToken).ConfigureAwait(false);
             await using var _ = topologyManager.ConfigureAwait(false);
 
+            foreach (var configureTopologyAction in _configureTopologyActions)
+            {
+                await configureTopologyAction(_serviceProvider, topologyManager).ConfigureAwait(false);
+            }
+
             foreach (var addressConfiguration in _addressConfigurations)
             {
                 await topologyManager.DeclareAddressAsync(addressConfiguration.Key, addressConfiguration.Value, cancellationToken).ConfigureAwait(false);

src/ArtemisNetClient.Extensions.DependencyInjection/Options/ActiveMqOptions.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 
 namespace ActiveMQ.Artemis.Client.Extensions.DependencyInjection
 {
@@ -8,8 +9,9 @@ internal class ActiveMqOptions
         public bool EnableQueueDeclaration { get; set; }
         public bool EnableAddressDeclaration { get; set; }
         public List<QueueConfiguration> QueueConfigurations { get; } = new List<QueueConfiguration>();
-        public Dictionary<string, HashSet<RoutingType>> AddressConfigurations { get; set; } = new Dictionary<string, HashSet<RoutingType>>();
+        public Dictionary<string, HashSet<RoutingType>> AddressConfigurations { get; } = new Dictionary<string, HashSet<RoutingType>>();
         public List<Action<IServiceProvider, ConnectionFactory>> ConnectionFactoryActions { get; } = new List<Action<IServiceProvider, ConnectionFactory>>();
+        public List<Func<IServiceProvider, ITopologyManager, Task>> ConfigureTopologyActions { get; } = new List<Func<IServiceProvider, ITopologyManager, Task>>();
         public List<Action<IServiceProvider, IConnection>> ConnectionActions { get; } = new List<Action<IServiceProvider, IConnection>>();
         public List<SendObserverRegistration> SendObserverRegistrations { get; } = new List<SendObserverRegistration>();
         public List<ReceiveObserverRegistration> ReceiveObserverRegistrations { get; } = new List<ReceiveObserverRegistration>();

test/ArtemisNetClient.Extensions.IntegrationTests/ConfigureTopologySpec.cs

@@ -0,0 +1,42 @@
+using System;
+using System.Threading.Tasks;
+using ActiveMQ.Artemis.Client.Extensions.DependencyInjection;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace ActiveMQ.Artemis.Client.Extensions.AspNetCore.IntegrationTests;
+
+public class ConfigureTopologySpec
+{
+    private readonly ITestOutputHelper _testOutputHelper;
+
+    public ConfigureTopologySpec(ITestOutputHelper testOutputHelper)
+    {
+        _testOutputHelper = testOutputHelper;
+    }
+
+    [Fact]
+    public async Task Should_configure_broker_topology_using_registered_delegate()
+    {
+        var addressName = Guid.NewGuid().ToString();
+        var queueName = Guid.NewGuid().ToString();
+
+        await using var testFixture = await TestFixture.CreateAsync(_testOutputHelper, activeMqBuilder =>
+        {
+            activeMqBuilder.ConfigureTopology(async (_, topologyManager) =>
+            {
+                await topologyManager.CreateQueueAsync(new QueueConfiguration
+                {
+                    Address = addressName,
+                    Name = queueName,
+                    AutoCreateAddress = true
+                });
+            });
+        });
+
+        var topologyManager = await testFixture.Connection.CreateTopologyManagerAsync(testFixture.CancellationToken);
+        var queueNames = await topologyManager.GetQueueNamesAsync(testFixture.CancellationToken);
+
+        Assert.Contains(queueName, queueNames);
+    }
+}