Dynamic topic discovery & reload signal for AzureBus

src/ChannelMediator.AzureBus/AzureServiceBusEntityManager.cs

@@ -1,5 +1,6 @@
 using System.Collections.Concurrent;
 
+using Azure.Messaging.ServiceBus;
 using Azure.Messaging.ServiceBus.Administration;
 
 using Microsoft.Extensions.Logging;
@@ -32,8 +33,8 @@ public AzureServiceBusEntityManager(ServiceBusAdministrationClient adminClient,
 	/// </summary>
 	/// <param name="topicName">The name of the topic.</param>
 	/// <param name="cancellationToken">Cancellation token.</param>
-	/// <returns>A task representing the asynchronous operation.</returns>
-	public async Task EnsureTopicExistsAsync(string topicName, CancellationToken cancellationToken = default)
+	/// <returns><c>true</c> if the topic was newly created; <c>false</c> if it already existed.</returns>
+	public async Task<bool> EnsureTopicExistsAsync(string topicName, CancellationToken cancellationToken = default)
 	{
 		ArgumentException.ThrowIfNullOrWhiteSpace(topicName);
 
@@ -43,9 +44,10 @@ public async Task EnsureTopicExistsAsync(string topicName, CancellationToken can
 		if (_queueOrTopicsExist.ContainsKey(topicName))
 		{
 			_logger.LogTrace("Topic '{TopicName}' already verified in cache.", topicName);
-			return;
+			return false;
 		}
 
+		bool created = false;
 		try
 		{
 			// Check if the topic exists in Azure Service Bus
@@ -62,9 +64,16 @@ public async Task EnsureTopicExistsAsync(string topicName, CancellationToken can
 				topicOptions.AuthorizationRules.Add(new SharedAccessAuthorizationRule("allClaims"
 					, new[] { AccessRights.Manage, AccessRights.Send, AccessRights.Listen }));
 
-				await _adminClient.CreateTopicAsync(topicOptions, cancellationToken);
-
-				_logger.LogInformation("Topic '{TopicName}' created.", topicName);
+				try
+				{
+					await _adminClient.CreateTopicAsync(topicOptions, cancellationToken);
+					_logger.LogInformation("Topic '{TopicName}' created.", topicName);
+					created = true;
+				}
+				catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists)
+				{
+					_logger.LogTrace("Topic '{TopicName}' was created concurrently by another process.", topicName);
+				}
 			}
 			else
 			{
@@ -79,6 +88,7 @@ public async Task EnsureTopicExistsAsync(string topicName, CancellationToken can
 
 		// Mark this topic as verified
 		_queueOrTopicsExist.TryAdd(topicName, true);
+		return created;
 	}
 
 	public async Task EnsureQueueExistsAsync(string queueName, CancellationToken cancellationToken = default)
@@ -108,9 +118,15 @@ public async Task EnsureQueueExistsAsync(string queueName, CancellationToken can
 				};
 				queueOptions.AuthorizationRules.Add(new SharedAccessAuthorizationRule("allClaims"
 					, new[] { AccessRights.Manage, AccessRights.Send, AccessRights.Listen }));
-				await _adminClient.CreateQueueAsync(queueOptions, cancellationToken);
-
-				_logger.LogInformation("Queue '{QueueName}' created.", queueName);
+				try
+				{
+					await _adminClient.CreateQueueAsync(queueOptions, cancellationToken);
+					_logger.LogInformation("Queue '{QueueName}' created.", queueName);
+				}
+				catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists)
+				{
+					_logger.LogTrace("Queue '{QueueName}' was created concurrently by another process.", queueName);
+				}
 			}
 			else
 			{
@@ -171,9 +187,15 @@ public async Task EnsureSubscriptionExistsAsync(
 					EnableBatchedOperations = true,
 				};
 
-				await _adminClient.CreateSubscriptionAsync(subscriptionOptions, cancellationToken);
-
-				_logger.LogInformation("Subscription '{SubscriptionKey}' created.", subscriptionKey);
+				try
+				{
+					await _adminClient.CreateSubscriptionAsync(subscriptionOptions, cancellationToken);
+					_logger.LogInformation("Subscription '{SubscriptionKey}' created.", subscriptionKey);
+				}
+				catch (ServiceBusException ex) when (ex.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists)
+				{
+					_logger.LogTrace("Subscription '{SubscriptionKey}' was created concurrently by another process.", subscriptionKey);
+				}
 			}
 			else
 			{

src/ChannelMediator.AzureBus/AzureServiceBusNameBuilder.cs

@@ -17,6 +17,15 @@ internal static string Build(string? prefix, string name)
         return NormalizeName(combined);
     }
 
+    /// <summary>
+    /// Builds the internal reload-topics topic name: <c>{prefix}reload-topics</c>.
+    /// This topic is created by design and used to signal readers that a new topic was created.
+    /// </summary>
+    internal static string BuildReloadTopicName(string? prefix)
+    {
+        return Build(prefix, "reload-topics");
+    }
+
     private static string NormalizeName(string value)
     {
         var lower = value.ToLowerInvariant();
@@ -37,6 +46,6 @@ private static string NormalizeName(string value)
         var result = builder.ToString().Trim('.', '-');
         result = result.Replace(".-", "-");
 
-		return result;
+        return result;
     }
 }

src/ChannelMediator.AzureBus/AzureServiceBusPublisher.cs

@@ -14,6 +14,7 @@ internal sealed class AzureServiceBusPublisher : IAzurePublisher, IAsyncDisposab
 	private readonly AzureServiceBusEntityManager _entityManager;
 	private readonly AzureServiceBusOptions _options;
 	private readonly ConcurrentDictionary<string, ServiceBusSender> _senders = new();
+	private readonly ConcurrentDictionary<string, Lazy<Task>> _topicSetupTasks = new();
 	private readonly JsonSerializerOptions _jsonOptions;
 	private bool _disposed;
 
@@ -43,14 +44,60 @@ public async Task Notify<TNotification>(TNotification notification, Cancellation
 		where TNotification : INotification
 	{
 		var queueOrTopicName = AzureServiceBusNameBuilder.Build(_options.Prefix, typeof(TNotification).Name);
-		await _entityManager.EnsureTopicExistsAsync(queueOrTopicName, cancellationToken);
+
+		// All concurrent callers for the same topic share a single setup task via Lazy<Task>.
+		// This ensures the topic and its subscription are fully created before any message is sent,
+		// even when multiple notifications of the same type are dispatched concurrently.
+		var lazySetup = _topicSetupTasks.GetOrAdd(
+			queueOrTopicName,
+			name => new Lazy<Task>(() => SetupTopicAsync(name, typeof(TNotification))));
+
+		await lazySetup.Value;
 
 		var sender = GetOrCreateSender(queueOrTopicName);
 		var message = CreateMessage(notification);
-
 		await sender.SendMessageAsync(message, cancellationToken);
 	}
 
+	/// <summary>
+	/// One-shot initialization for a topic: creates the topic if needed, ensures the subscription
+	/// exists before any message is sent, then broadcasts the reload signal to readers.
+	/// Uses <see cref="CancellationToken.None"/> so that a single caller's cancellation
+	/// does not abort setup for all concurrent callers.
+	/// </summary>
+	private async Task SetupTopicAsync(string topicName, Type notificationType)
+	{
+		var isNewTopic = await _entityManager.EnsureTopicExistsAsync(topicName);
+		if (!isNewTopic)
+		{
+			return;
+		}
+
+		await _entityManager.EnsureSubscriptionExistsAsync(
+			topicName,
+			_options.TopicSubscriberName,
+			notificationType);
+
+		await SendReloadSignalAsync(topicName, CancellationToken.None);
+	}
+
+	/// <summary>
+	/// Sends a reload signal to the internal <c>{prefix}reload-topics</c> topic so that readers
+	/// can discover and subscribe to the newly created topic.
+	/// </summary>
+	private async Task SendReloadSignalAsync(string newTopicName, CancellationToken cancellationToken)
+	{
+		var reloadTopicName = AzureServiceBusNameBuilder.BuildReloadTopicName(_options.Prefix);
+
+		// Ensure the reload topic exists — it is normally created by readers at startup,
+		// but the writer may start before any reader.
+		await _entityManager.EnsureTopicExistsAsync(reloadTopicName, cancellationToken);
+
+		var sender = GetOrCreateSender(reloadTopicName);
+		var signal = new ServiceBusMessage(newTopicName) { ContentType = "text/plain" };
+		await sender.SendMessageAsync(signal, cancellationToken);
+	}
+
 	/// <inheritdoc />
 	public async Task EnqueueRequest<R>(R request, CancellationToken cancellationToken = default)
 		where R : IRequest

src/ChannelMediator.AzureBus/ChannelMediator.AzureBus.csproj

@@ -4,14 +4,14 @@
    <TargetFrameworks>net10.0;net9.0;net8.0</TargetFrameworks>
 		<ImplicitUsings>enable</ImplicitUsings>
 		<Nullable>enable</Nullable>
-   <GenerateDocumentationFile>true</GenerateDocumentationFile>
+   		<GenerateDocumentationFile>true</GenerateDocumentationFile>
 		<IncludeSymbols>true</IncludeSymbols>
 		<SymbolPackageFormat>snupkg</SymbolPackageFormat>
 		<EmbedUntrackedSources>true</EmbedUntrackedSources>
-		<Version>1.1.12.0</Version>
-    <Authors>Appliman</Authors>
+		<Version>1.1.13.0</Version>
+    	<Authors>Appliman</Authors>
 		<PackageProjectUrl>https://github.com/appliman/channelmediator</PackageProjectUrl>
-    <RepositoryUrl>https://github.com/appliman/channelmediator</RepositoryUrl>
+    	<RepositoryUrl>https://github.com/appliman/channelmediator</RepositoryUrl>
 		<RepositoryType>git</RepositoryType>
 		<PublishRepositoryUrl>true</PublishRepositoryUrl>
 		<PackageTags>ChannelMediator;AzureServiceBus;ServiceBus;Messaging;Queues;Topics</PackageTags>

src/ChannelMediator.AzureBus/TopicSubscriptionReader.cs

@@ -165,7 +165,7 @@ private async Task DispatchNotificationAsync(object notification, CancellationTo
 		var mediator = _serviceProvider.GetRequiredService<IMediator>();
 
 		_logger.LogTrace("Publishing notification of type {NotificationType} from topic {Topic}/{Subscription}.", notification.GetType().FullName, _options.TopicName, _options.SubscriptionName);
-		await mediator.Publish((INotification)notification, cancellationToken);
+		await mediator.Publish((dynamic)notification, cancellationToken);
 	}
 
 	private Task ProcessErrorAsync(ProcessErrorEventArgs args)

src/ChannelMediator.AzureBus/TopicSubscriptionReadersHostedService.cs

@@ -14,6 +14,9 @@ internal sealed class TopicSubscriptionReadersHostedService : IHostedService, IA
 {
 	private readonly IServiceProvider _serviceProvider;
 	private readonly List<TopicSubscriptionReader> _readers = [];
+	private readonly SemaphoreSlim _refreshLock = new(1, 1);
+	private readonly HashSet<string> _subscribedTopics = new(StringComparer.OrdinalIgnoreCase);
+	private ServiceBusProcessor? _reloadProcessor;
 	private bool _disposed;
 	private readonly ILogger _logger;
 
@@ -43,7 +46,6 @@ public async Task StartAsync(CancellationToken cancellationToken)
 
 		if (!options.SubscribeToAllTopics)
 		{
-
 			foreach (var readerOptions in TopicSubscriptionReaderRegistry.GetAll())
 			{
 				var reader = ActivatorUtilities.CreateInstance<TopicSubscriptionReader>(_serviceProvider, client, entityManager, readerOptions, _serviceProvider);
@@ -53,18 +55,99 @@ public async Task StartAsync(CancellationToken cancellationToken)
 		}
 		else
 		{
-			var subscriptionName = options.TopicSubscriberName.ToLowerInvariant();
 			var adminClient = _serviceProvider.GetRequiredService<ServiceBusAdministrationClient>();
 
-			await foreach (var topic in adminClient.GetTopicsAsync())
+			// Always start the internal reload-topics reader first — created by design.
+			await StartReloadTopicsReaderAsync(options, client, entityManager, adminClient, cancellationToken);
+
+			// Initial scan: subscribe to all existing topics matching the prefix.
+			await RefreshSubscriptionsAsync(options, client, entityManager, adminClient, cancellationToken);
+		}
+	}
+
+	/// <summary>
+	/// Creates the internal <c>{prefix}reload-topics</c> topic and starts a processor on it.
+	/// When a message arrives (sent by the publisher after creating a new topic), triggers
+	/// a full subscription refresh so the reader picks up the new topic immediately.
+	/// </summary>
+	private async Task StartReloadTopicsReaderAsync(
+		AzureServiceBusOptions options,
+		ServiceBusClient client,
+		AzureServiceBusEntityManager entityManager,
+		ServiceBusAdministrationClient adminClient,
+		CancellationToken cancellationToken)
+	{
+		var reloadTopicName = AzureServiceBusNameBuilder.BuildReloadTopicName(options.Prefix);
+		var subscriptionName = options.TopicSubscriberName.ToLowerInvariant();
+
+		await entityManager.EnsureTopicExistsAsync(reloadTopicName, cancellationToken);
+		await entityManager.EnsureSubscriptionExistsAsync(reloadTopicName, subscriptionName, typeof(INotification), cancellationToken);
+
+		_logger.LogInformation("[reload-topics] Listening on internal topic '{ReloadTopic}' / subscription '{Subscription}'.", reloadTopicName, subscriptionName);
+
+		var processorOptions = new ServiceBusProcessorOptions
+		{
+			MaxConcurrentCalls = 1,
+			AutoCompleteMessages = false,
+			ReceiveMode = ServiceBusReceiveMode.ReceiveAndDelete
+		};
+
+		_reloadProcessor = client.CreateProcessor(reloadTopicName, subscriptionName, processorOptions);
+
+		_reloadProcessor.ProcessMessageAsync += async args =>
+		{
+			var newTopicName = args.Message.Body.ToString();
+			_logger.LogInformation("[reload-topics] Signal received: topic '{NewTopic}' was created. Refreshing subscriptions...", newTopicName);
+			await RefreshSubscriptionsAsync(options, client, entityManager, adminClient, args.CancellationToken);
+		};
+
+		_reloadProcessor.ProcessErrorAsync += args =>
+		{
+			_logger.LogError(args.Exception, "[reload-topics] Error processing reload signal.");
+			return Task.CompletedTask;
+		};
+
+		await _reloadProcessor.StartProcessingAsync(cancellationToken);
+	}
+
+	/// <summary>
+	/// Scans all Azure Service Bus topics matching <c>{prefix}*</c> (excluding the reload topic itself)
+	/// and subscribes to any that are not yet tracked. Safe to call concurrently — serialised by a lock.
+	/// </summary>
+	private async Task RefreshSubscriptionsAsync(
+		AzureServiceBusOptions options,
+		ServiceBusClient client,
+		AzureServiceBusEntityManager entityManager,
+		ServiceBusAdministrationClient adminClient,
+		CancellationToken cancellationToken)
+	{
+		await _refreshLock.WaitAsync(cancellationToken);
+		try
+		{
+			var subscriptionName = options.TopicSubscriberName.ToLowerInvariant();
+			var reloadTopicName = AzureServiceBusNameBuilder.BuildReloadTopicName(options.Prefix);
+
+			await foreach (var topic in adminClient.GetTopicsAsync(cancellationToken))
 			{
-				_logger.LogDebug("Checking topic {TopicName} for subscription reader creation. status : {Status}.", topic.Name, topic.Status);
 				if (!topic.Name.StartsWith(options.Prefix, StringComparison.OrdinalIgnoreCase))
 				{
 					continue;
 				}
 
-				await entityManager.EnsureSubscriptionExistsAsync(topic.Name, subscriptionName, typeof(INotification));
+				// Skip the internal reload-topics topic — it is managed separately.
+				if (string.Equals(topic.Name, reloadTopicName, StringComparison.OrdinalIgnoreCase))
+				{
+					continue;
+				}
+
+				if (_subscribedTopics.Contains(topic.Name))
+				{
+					continue;
+				}
+
+				_logger.LogDebug("Subscribing to topic '{TopicName}' (discovered via refresh).", topic.Name);
+
+				await entityManager.EnsureSubscriptionExistsAsync(topic.Name, subscriptionName, typeof(INotification), cancellationToken);
 
 				var readerOptions = new TopicSubscriptionReaderOptions
 				{
@@ -79,13 +162,25 @@ public async Task StartAsync(CancellationToken cancellationToken)
 				var reader = ActivatorUtilities.CreateInstance<TopicSubscriptionReader>(_serviceProvider, client, entityManager, readerOptions, _serviceProvider);
 				_readers.Add(reader);
 				await reader.StartAsync(cancellationToken);
+
+				_subscribedTopics.Add(topic.Name);
+				_logger.LogInformation("Now listening on topic '{TopicName}'.", topic.Name);
 			}
 		}
+		finally
+		{
+			_refreshLock.Release();
+		}
 	}
 
 	/// <inheritdoc />
 	public async Task StopAsync(CancellationToken cancellationToken)
 	{
+		if (_reloadProcessor is not null)
+		{
+			await _reloadProcessor.StopProcessingAsync(cancellationToken);
+		}
+
 		foreach (var reader in _readers)
 		{
 			await reader.StopAsync(cancellationToken);
@@ -102,11 +197,17 @@ public async ValueTask DisposeAsync()
 
 		_disposed = true;
 
+		if (_reloadProcessor is not null)
+		{
+			await _reloadProcessor.DisposeAsync();
+		}
+
 		foreach (var reader in _readers)
 		{
 			await reader.DisposeAsync();
 		}
 
 		_readers.Clear();
+		_refreshLock.Dispose();
 	}
 }

src/Samples/AzureBusReaderSampleConsole/OrderShippedHandler.cs

@@ -0,0 +1,14 @@
+using ChannelMediator;
+
+using ChannelMediatorSampleShared;
+
+namespace AzureBusReaderSampleReaderConsole;
+
+public sealed class OrderShippedHandler : INotificationHandler<OrderShippedNotification>
+{
+    public async Task Handle(OrderShippedNotification notification, CancellationToken cancellationToken)
+    {
+        await Task.Delay(20, cancellationToken);
+        Console.WriteLine($"[ORDER] Order {notification.OrderId} shipped to {notification.Destination} at {notification.ShippedAt:u}");
+    }
+}