[Host.AzureServiceBus] Create topic/queue dynamically when path param uses a non-default value

Signed-off-by: Tomasz Maruszak <[email protected]>

Author: zarusz

src/Host.Plugin.Properties.xml

@@ -4,7 +4,7 @@
   <Import Project="Common.NuGet.Properties.xml" />
 
   <PropertyGroup>
-    <Version>3.3.1-rc100</Version>
+    <Version>3.3.1-rc101</Version>
   </PropertyGroup>
 
 </Project>

src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs

@@ -7,12 +7,12 @@ public class ServiceBusMessageBus : MessageBusBase<ServiceBusMessageBusSettings>
     private readonly ILogger _logger;
     private ServiceBusClient _client;
     private SafeDictionaryWrapper<string, ServiceBusSender> _producerByPath;
+    private ServiceBusTopologyService _topologyService;
 
     public ServiceBusMessageBus(MessageBusSettings settings, ServiceBusMessageBusSettings providerSettings)
         : base(settings, providerSettings)
     {
         _logger = LoggerFactory.CreateLogger<ServiceBusMessageBus>();
-
         OnBuildProvider();
     }
 
@@ -46,9 +46,7 @@ protected override async ValueTask DisposeAsyncCore()
     public override async Task ProvisionTopology()
     {
         await base.ProvisionTopology();
-
-        var provisioningService = new ServiceBusTopologyService(LoggerFactory.CreateLogger<ServiceBusTopologyService>(), Settings, ProviderSettings);
-        await provisioningService.ProvisionTopology(); // provisioning happens asynchronously
+        await _topologyService?.ProvisionTopology(); // provisioning happens asynchronously
     }
 
     #region Overrides of MessageBusBase
@@ -59,6 +57,7 @@ protected override void Build()
 
         if (ProviderSettings.TopologyProvisioning?.Enabled ?? false)
         {
+            _topologyService = new ServiceBusTopologyService(LoggerFactory.CreateLogger<ServiceBusTopologyService>(), Settings, ProviderSettings);
             InitTaskList.Add(ProvisionTopology, CancellationToken);
         }
 
@@ -133,15 +132,48 @@ public override async Task ProduceToTransport(object message, Type messageType,
         {
             var transportMessage = GetTransportMessage(message, messageType, messageHeaders, path);
             var senderClient = _producerByPath.GetOrAdd(path);
-            await senderClient.SendMessageAsync(transportMessage, cancellationToken).ConfigureAwait(false);
-            _logger.LogDebug("Delivered item {Message} of type {MessageType} to {Path}", message, messageType?.Name, path);
+
+            try
+            {
+                await senderClient.SendMessageAsync(transportMessage, cancellationToken).ConfigureAwait(false);
+            }
+            catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityNotFound && _topologyService != null)
+            {
+                await EnsurePathExists(messageType, path);
+                // Resend messages after the path has been created
+                await senderClient.SendMessageAsync(transportMessage, cancellationToken).ConfigureAwait(false);
+            }
+
+            _logger.LogDebug("Delivered message {Message} of type {MessageType} to {Path}", message, messageType?.Name, path);
         }
         catch (Exception ex) when (ex is not ProducerMessageBusException && ex is not TaskCanceledException)
         {
             throw new ProducerMessageBusException(GetProducerErrorMessage(path, message, messageType, ex), ex);
         }
     }
 
+    /// <summary>
+    /// When the topic or queue does not exist, we can try to create it. 
+    /// This happens in cases where the path is dynamically set upon publish/send (different from default path)
+    /// </summary>
+    /// <param name="messageType"></param>
+    /// <param name="path">topic or queue</param>
+    /// <returns></returns>
+    private async Task EnsurePathExists(Type messageType, string path)
+    {
+        var producerSettings = GetProducerSettings(messageType);
+        if (producerSettings.PathKind == PathKind.Topic)
+        {
+            _logger.LogInformation("Topic {Path} does not exist, trying to create it", path);
+            await _topologyService.TryCreateTopic(path, ProviderSettings.TopologyProvisioning.CanProducerCreateTopic);
+        }
+        else
+        {
+            _logger.LogInformation("Queue {Path} does not exist, trying to create it", path);
+            await _topologyService.TryCreateQueue(path, ProviderSettings.TopologyProvisioning.CanProducerCreateQueue);
+        }
+    }
+
     public override async Task<ProduceToTransportBulkResult<T>> ProduceToTransportBulk<T>(IReadOnlyCollection<T> envelopes, string path, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
         AssertActive();
@@ -150,7 +182,21 @@ Task SendBatchAsync(ServiceBusSender senderClient, ServiceBusMessageBatch batch,
             Retry.WithDelay(
                 operation: async cancellationToken =>
                 {
-                    await senderClient.SendMessagesAsync(batch, cancellationToken).ConfigureAwait(false);
+                    try
+                    {
+                        await senderClient.SendMessagesAsync(batch, cancellationToken).ConfigureAwait(false);
+                    }
+                    catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityNotFound && _topologyService != null)
+                    {
+                        var messageType = envelopes.FirstOrDefault().MessageType;
+                        if (messageType != null)
+                        {
+                            await EnsurePathExists(messageType, path);
+                            // Resend messages after the path has been created
+                            await senderClient.SendMessagesAsync(batch, cancellationToken).ConfigureAwait(false);
+                        }
+                    }
+
                     _logger.LogDebug("Batch of {BatchSize} message(s) dispatched to {Path} ({SizeInBytes} bytes)", batch.Count, path, batch.SizeInBytes);
                 },
                 shouldRetry: (exception, attempt) =>

src/SlimMessageBus.Host.AzureServiceBus/ServiceBusTopologyService.cs

@@ -16,7 +16,7 @@ public ServiceBusTopologyService(ILogger<ServiceBusTopologyService> logger, Mess
     }
 
     [Flags]
-    private enum TopologyCreationStatus
+    internal enum TopologyCreationStatus
     {
         None = 0,
         NotExists = 1,
@@ -51,7 +51,7 @@ private static async Task<T> SwallowExceptionIfMessagingEntityNotFound<T>(Func<T
         }
     }
 
-    private Task<TopologyCreationStatus> TryCreateQueue(string path, bool canCreate, Action<CreateQueueOptions> action) => SwallowExceptionIfEntityExists(async () =>
+    internal Task<TopologyCreationStatus> TryCreateQueue(string path, bool canCreate, Action<CreateQueueOptions> action = null) => SwallowExceptionIfEntityExists(async () =>
     {
         if (await _adminClient.QueueExistsAsync(path)) return TopologyCreationStatus.Exists;
 
@@ -71,7 +71,7 @@ private Task<TopologyCreationStatus> TryCreateQueue(string path, bool canCreate,
         return TopologyCreationStatus.Exists | TopologyCreationStatus.Created;
     });
 
-    private Task<TopologyCreationStatus> TryCreateTopic(string path, bool canCreate, Action<CreateTopicOptions> action) => SwallowExceptionIfEntityExists(async () =>
+    internal Task<TopologyCreationStatus> TryCreateTopic(string path, bool canCreate, Action<CreateTopicOptions> action = null) => SwallowExceptionIfEntityExists(async () =>
     {
         if (await _adminClient.TopicExistsAsync(path)) return TopologyCreationStatus.Exists;
 

src/Tests/SlimMessageBus.Host.AzureServiceBus.Test/ServiceBusMessageBusTests.cs

@@ -4,6 +4,7 @@
 using System.Globalization;
 
 using Azure.Messaging.ServiceBus;
+using Azure.Messaging.ServiceBus.Administration;
 
 using SlimMessageBus.Host;
 using SlimMessageBus.Host.Collections;
@@ -47,6 +48,11 @@ public ServiceBusMessageBusTests()
                 SenderMockByPath.Add(path, m);
                 return m.Object;
             },
+            AdminClientFactory = (_, _) =>
+            {
+                var adminClient = new Mock<ServiceBusAdministrationClient>();
+                return adminClient.Object;
+            },
             TopologyProvisioning = new ServiceBusTopologySettings
             {
                 Enabled = false