diff --git a/src/ChannelMediator.AzureBus/AzureServiceBusEntityManager.cs b/src/ChannelMediator.AzureBus/AzureServiceBusEntityManager.cs index 6720379..3bad17a 100644 --- a/src/ChannelMediator.AzureBus/AzureServiceBusEntityManager.cs +++ b/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, /// /// The name of the topic. /// Cancellation token. - /// A task representing the asynchronous operation. - public async Task EnsureTopicExistsAsync(string topicName, CancellationToken cancellationToken = default) + /// true if the topic was newly created; false if it already existed. + public async Task 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 { diff --git a/src/ChannelMediator.AzureBus/AzureServiceBusNameBuilder.cs b/src/ChannelMediator.AzureBus/AzureServiceBusNameBuilder.cs index b760989..6e8470b 100644 --- a/src/ChannelMediator.AzureBus/AzureServiceBusNameBuilder.cs +++ b/src/ChannelMediator.AzureBus/AzureServiceBusNameBuilder.cs @@ -17,6 +17,15 @@ internal static string Build(string? prefix, string name) return NormalizeName(combined); } + /// + /// Builds the internal reload-topics topic name: {prefix}reload-topics. + /// This topic is created by design and used to signal readers that a new topic was created. + /// + 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; } } diff --git a/src/ChannelMediator.AzureBus/AzureServiceBusPublisher.cs b/src/ChannelMediator.AzureBus/AzureServiceBusPublisher.cs index 7a2bb41..4f443a3 100644 --- a/src/ChannelMediator.AzureBus/AzureServiceBusPublisher.cs +++ b/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 _senders = new(); + private readonly ConcurrentDictionary> _topicSetupTasks = new(); private readonly JsonSerializerOptions _jsonOptions; private bool _disposed; @@ -43,14 +44,60 @@ public async Task Notify(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. + // 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(() => SetupTopicAsync(name, typeof(TNotification)))); + + await lazySetup.Value; var sender = GetOrCreateSender(queueOrTopicName); var message = CreateMessage(notification); - await sender.SendMessageAsync(message, cancellationToken); } + /// + /// 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 so that a single caller's cancellation + /// does not abort setup for all concurrent callers. + /// + 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); + } + + /// + /// Sends a reload signal to the internal {prefix}reload-topics topic so that readers + /// can discover and subscribe to the newly created topic. + /// + 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); + } + /// public async Task EnqueueRequest(R request, CancellationToken cancellationToken = default) where R : IRequest diff --git a/src/ChannelMediator.AzureBus/ChannelMediator.AzureBus.csproj b/src/ChannelMediator.AzureBus/ChannelMediator.AzureBus.csproj index 7b0a4b5..2bf5857 100644 --- a/src/ChannelMediator.AzureBus/ChannelMediator.AzureBus.csproj +++ b/src/ChannelMediator.AzureBus/ChannelMediator.AzureBus.csproj @@ -4,14 +4,14 @@ net10.0;net9.0;net8.0 enable enable - true + true true snupkg true - 1.1.12.0 - Appliman + 1.1.13.0 + Appliman https://github.com/appliman/channelmediator - https://github.com/appliman/channelmediator + https://github.com/appliman/channelmediator git true ChannelMediator;AzureServiceBus;ServiceBus;Messaging;Queues;Topics diff --git a/src/ChannelMediator.AzureBus/TopicSubscriptionReader.cs b/src/ChannelMediator.AzureBus/TopicSubscriptionReader.cs index 706718d..878dcf7 100644 --- a/src/ChannelMediator.AzureBus/TopicSubscriptionReader.cs +++ b/src/ChannelMediator.AzureBus/TopicSubscriptionReader.cs @@ -165,7 +165,7 @@ private async Task DispatchNotificationAsync(object notification, CancellationTo var mediator = _serviceProvider.GetRequiredService(); _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) diff --git a/src/ChannelMediator.AzureBus/TopicSubscriptionReadersHostedService.cs b/src/ChannelMediator.AzureBus/TopicSubscriptionReadersHostedService.cs index cb60196..3deed6d 100644 --- a/src/ChannelMediator.AzureBus/TopicSubscriptionReadersHostedService.cs +++ b/src/ChannelMediator.AzureBus/TopicSubscriptionReadersHostedService.cs @@ -14,6 +14,9 @@ internal sealed class TopicSubscriptionReadersHostedService : IHostedService, IA { private readonly IServiceProvider _serviceProvider; private readonly List _readers = []; + private readonly SemaphoreSlim _refreshLock = new(1, 1); + private readonly HashSet _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(_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(); - 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); + } + } + + /// + /// Creates the internal {prefix}reload-topics 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. + /// + 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); + } + + /// + /// Scans all Azure Service Bus topics matching {prefix}* (excluding the reload topic itself) + /// and subscribes to any that are not yet tracked. Safe to call concurrently — serialised by a lock. + /// + 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(_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(); + } } /// 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(); } } diff --git a/src/Samples/AzureBusReaderSampleConsole/OrderShippedHandler.cs b/src/Samples/AzureBusReaderSampleConsole/OrderShippedHandler.cs new file mode 100644 index 0000000..628ebb3 --- /dev/null +++ b/src/Samples/AzureBusReaderSampleConsole/OrderShippedHandler.cs @@ -0,0 +1,14 @@ +using ChannelMediator; + +using ChannelMediatorSampleShared; + +namespace AzureBusReaderSampleReaderConsole; + +public sealed class OrderShippedHandler : INotificationHandler +{ + 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}"); + } +} diff --git a/src/Samples/AzureBusReaderSampleConsole/Program.cs b/src/Samples/AzureBusReaderSampleConsole/Program.cs index bcd1475..1aeaa88 100644 --- a/src/Samples/AzureBusReaderSampleConsole/Program.cs +++ b/src/Samples/AzureBusReaderSampleConsole/Program.cs @@ -9,6 +9,7 @@ using Microsoft.Extensions.Hosting; var host = Host.CreateDefaultBuilder(args) + .UseEnvironment(Environments.Development) .ConfigureServices((context, services) => { var connectionString = context.Configuration.GetConnectionString("AzureBusConnectionString"); diff --git a/src/Samples/AzureBusWriterSampleConsole/Program.cs b/src/Samples/AzureBusWriterSampleConsole/Program.cs index a0ad443..2f53516 100644 --- a/src/Samples/AzureBusWriterSampleConsole/Program.cs +++ b/src/Samples/AzureBusWriterSampleConsole/Program.cs @@ -10,7 +10,7 @@ using Microsoft.Extensions.Hosting; var host = Host.CreateDefaultBuilder(args); - +host.UseEnvironment(Environments.Development); host.ConfigureServices((context, services) => { var connectionString = context.Configuration["ConnectionStrings:AzureBusConnectionString"]; @@ -35,7 +35,23 @@ var mediator = app.Services.GetRequiredService(); await mediator.EnqueueRequest(new MyRequest("enqueue-test")); -await mediator.Notify(new ProductAddedNotification("p01",10, 100)); +await mediator.Notify(new ProductAddedNotification("p01", 10, 100)); + +// Publish multiple OrderShippedNotification messages and verify delivery via logs +var orders = new[] +{ + new OrderShippedNotification("ORD-001", "Paris, France", DateTimeOffset.UtcNow), + new OrderShippedNotification("ORD-002", "Lyon, France", DateTimeOffset.UtcNow.AddSeconds(1)), + new OrderShippedNotification("ORD-003", "Marseille, France", DateTimeOffset.UtcNow.AddSeconds(2)), +}; + +foreach (var order in orders) +{ + await mediator.Notify(order); + Console.WriteLine($"[WRITER] Sent OrderShippedNotification for order {order.OrderId} → {order.Destination}"); +} -Console.WriteLine("Notification published. Press any key to exit."); +Console.WriteLine(); +Console.WriteLine("All notifications published. Start AzureBusReaderSampleConsole to verify the topic receives the messages."); +Console.WriteLine("Press any key to exit."); Console.ReadLine(); \ No newline at end of file diff --git a/src/Samples/ChannelMediatorSampleShared/OrderShippedNotification.cs b/src/Samples/ChannelMediatorSampleShared/OrderShippedNotification.cs new file mode 100644 index 0000000..9400c38 --- /dev/null +++ b/src/Samples/ChannelMediatorSampleShared/OrderShippedNotification.cs @@ -0,0 +1,5 @@ +using ChannelMediator; + +namespace ChannelMediatorSampleShared; + +public record OrderShippedNotification(string OrderId, string Destination, DateTimeOffset ShippedAt) : INotification;