Merge pull request #1787 from dosper7/azure-service-bus-emulator-support

add azure service bus AutoProvision property

Author: demorgi

README.md

@@ -287,6 +287,25 @@ services.AddCap(x =>
 });
 ```
 
+### Azure Service Bus Emulator Support
+
+The [Azure Service Bus Emulator](https://learn.microsoft.com/en-us/azure/service-bus-messaging/overview-emulator) uses separate ports for AMQP messaging (5672) and the HTTP Admin API (5300). Because CAP uses a single connection string for both the `ServiceBusClient` (AMQP) and the `ServiceBusAdministrationClient` (HTTP), it cannot target both ports simultaneously.
+
+To work around this, set `AutoProvision` to `false` to skip automatic creation of topics, subscriptions, and rules via the Admin API. You must pre-create the required entities in the emulator's configuration instead.
+
+```csharp
+services.AddCap(x =>
+{
+    x.UseAzureServiceBus(opt =>
+    {
+        opt.ConnectionString = "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;";
+        opt.AutoProvision = false;
+    });
+});
+```
+
+> **Note:** When `AutoProvision` is `false`, topics, subscriptions, and subscription filter rules must already exist before the application starts. This option is also useful when entities are managed externally (e.g., via Infrastructure as Code).
+
 ## Dashboard
 
 CAP provides a real-time dashboard to view sent and received messages and their status.
@@ -307,4 +326,4 @@ We welcome contributions! Participating in discussions, reporting issues, and su
 
 ### License
 
-CAP is licensed under the [MIT License](https://github.com/dotnetcore/CAP/blob/master/LICENSE.txt).
+CAP is licensed under the [MIT License](https://github.com/dotnetcore/CAP/blob/master/LICENSE.txt).

src/DotNetCore.CAP.AzureServiceBus/AzureServiceBusConsumerClient.cs

@@ -57,7 +57,10 @@ public async Task SubscribeAsync(IEnumerable<string> topics)
 
         await ConnectAsync();
 
-        topics = topics.Concat(_asbOptions!.SQLFilters?.Select(o => o.Key) ?? Enumerable.Empty<string>());
+        if (!_asbOptions.AutoProvision) 
+            return;
+
+        topics = topics.Concat(_asbOptions!.SQLFilters?.Select(o => o.Key) ?? []);
 
         var allRules = _administrationClient!.GetRulesAsync(_asbOptions!.TopicPath, _subscriptionName).ToEnumerable();
         var allRuleNames = allRules.Select(o => o.Name);
@@ -88,21 +91,21 @@ public async Task SubscribeAsync(IEnumerable<string> topics)
                 currentRuleToAdd = correlationRule;
             }
 
-           await _administrationClient.CreateRuleAsync(_asbOptions.TopicPath, _subscriptionName,
-                new CreateRuleOptions
-                {
-                    Name = newRule,
-                    Filter = currentRuleToAdd
-                });
+            await _administrationClient.CreateRuleAsync(_asbOptions.TopicPath, _subscriptionName,
+                 new CreateRuleOptions
+                 {
+                     Name = newRule,
+                     Filter = currentRuleToAdd
+                 });
 
-            _logger.LogInformation($"Azure Service Bus add rule: {newRule}");
+            _logger.LogInformation("Azure Service Bus add rule: {RuleName}", newRule);
         }
 
         foreach (var oldRule in allRuleNames.Except(topics))
         {
             await _administrationClient.DeleteRuleAsync(_asbOptions.TopicPath, _subscriptionName, oldRule);
 
-            _logger.LogInformation($"Azure Service Bus remove rule: {oldRule}");
+            _logger.LogInformation("Azure Service Bus remove rule: {RuleName}", oldRule);
         }
     }
 
@@ -141,7 +144,7 @@ public async Task RejectAsync(object? sender)
 
     public async ValueTask DisposeAsync()
     {
-        if (!_serviceBusProcessor!.IsProcessing) 
+        if (!_serviceBusProcessor!.IsProcessing)
             await _serviceBusProcessor.DisposeAsync();
     }
 
@@ -196,49 +199,55 @@ public async Task ConnectAsync()
         {
             if (_serviceBusProcessor == null)
             {
-                if (_asbOptions.TokenCredential != null)
-                {
-                    _administrationClient =
-                        new ServiceBusAdministrationClient(_asbOptions.Namespace, _asbOptions.TokenCredential);
-                    _serviceBusClient = new ServiceBusClient(_asbOptions.Namespace, _asbOptions.TokenCredential);
-                }
-                else
-                {
-                    _administrationClient = new ServiceBusAdministrationClient(_asbOptions.ConnectionString);
-                    _serviceBusClient = new ServiceBusClient(_asbOptions.ConnectionString);
-                }
 
-                var topicConfigs =
-                    _asbOptions.CustomProducers.Select(producer =>
-                            (topicPaths: producer.TopicPath, subscribe: producer.CreateSubscription))
-                        .Append((topicPaths: _asbOptions.TopicPath, subscribe: true))
-                        .GroupBy(n => n.topicPaths, StringComparer.OrdinalIgnoreCase)
-                        .Select(n => (topicPaths: n.Key, subscribe: n.Max(o => o.subscribe)));
+                _serviceBusClient = _asbOptions.TokenCredential is not null ?
+                    new ServiceBusClient(_asbOptions.Namespace, _asbOptions.TokenCredential) :
+                    new ServiceBusClient(_asbOptions.ConnectionString);
 
-                foreach (var (topicPath, subscribe) in topicConfigs)
+                if (_asbOptions.AutoProvision)
                 {
-                    if (!await _administrationClient.TopicExistsAsync(topicPath))
+                    if (_asbOptions.TokenCredential != null)
                     {
-                        await _administrationClient.CreateTopicAsync(topicPath);
-                        _logger.LogInformation($"Azure Service Bus created topic: {topicPath}");
+                        _administrationClient =
+                            new ServiceBusAdministrationClient(_asbOptions.Namespace, _asbOptions.TokenCredential);
                     }
-
-                    if (subscribe && !await _administrationClient.SubscriptionExistsAsync(topicPath, _subscriptionName))
+                    else
                     {
-                        var subscriptionDescription =
-                            new CreateSubscriptionOptions(topicPath, _subscriptionName)
-                            {
-                                RequiresSession = _asbOptions.EnableSessions,
-                                AutoDeleteOnIdle = _asbOptions.SubscriptionAutoDeleteOnIdle,
-                                LockDuration = _asbOptions.SubscriptionMessageLockDuration,
-                                DefaultMessageTimeToLive = _asbOptions.SubscriptionDefaultMessageTimeToLive,
-                                MaxDeliveryCount = _asbOptions.SubscriptionMaxDeliveryCount,
-                            };
+                        _administrationClient = new ServiceBusAdministrationClient(_asbOptions.ConnectionString);
+                    }
 
-                        await _administrationClient.CreateSubscriptionAsync(subscriptionDescription);
+                    var topicConfigs =
+                        _asbOptions.CustomProducers.Select(producer =>
+                                (topicPaths: producer.TopicPath, subscribe: producer.CreateSubscription))
+                            .Append((topicPaths: _asbOptions.TopicPath, subscribe: true))
+                            .GroupBy(n => n.topicPaths, StringComparer.OrdinalIgnoreCase)
+                            .Select(n => (topicPaths: n.Key, subscribe: n.Max(o => o.subscribe)));
 
-                        _logger.LogInformation(
-                            $"Azure Service Bus topic {topicPath} created subscription: {_subscriptionName}");
+                    foreach (var (topicPath, subscribe) in topicConfigs)
+                    {
+                        if (!await _administrationClient.TopicExistsAsync(topicPath))
+                        {
+                            await _administrationClient.CreateTopicAsync(topicPath);
+                            _logger.LogInformation("Azure Service Bus created topic: {TopicPath}", topicPath);
+                        }
+
+                        if (subscribe && !await _administrationClient.SubscriptionExistsAsync(topicPath, _subscriptionName))
+                        {
+                            var subscriptionDescription =
+                                new CreateSubscriptionOptions(topicPath, _subscriptionName)
+                                {
+                                    RequiresSession = _asbOptions.EnableSessions,
+                                    AutoDeleteOnIdle = _asbOptions.SubscriptionAutoDeleteOnIdle,
+                                    LockDuration = _asbOptions.SubscriptionMessageLockDuration,
+                                    DefaultMessageTimeToLive = _asbOptions.SubscriptionDefaultMessageTimeToLive,
+                                    MaxDeliveryCount = _asbOptions.SubscriptionMaxDeliveryCount,
+                                };
+
+                            await _administrationClient.CreateSubscriptionAsync(subscriptionDescription);
+
+                            _logger.LogInformation(
+                                $"Azure Service Bus topic {topicPath} created subscription: {_subscriptionName}");
+                        }
                     }
                 }
 
@@ -295,39 +304,5 @@ private TransportMessage ConvertMessage(ServiceBusReceivedMessage message)
         return new TransportMessage(headers, message.Body);
     }
 
-    private static void CheckValidSubscriptionName(string subscriptionName)
-    {
-        const string pathDelimiter = @"/";
-        const int ruleNameMaximumLength = 50;
-        char[] invalidEntityPathCharacters = { '@', '?', '#', '*' };
-
-        if (string.IsNullOrWhiteSpace(subscriptionName)) throw new ArgumentNullException(subscriptionName);
-
-        // and "\" will be converted to "/" on the REST path anyway. Gateway/REST do not
-        // have to worry about the begin/end slash problem, so this is purely a client side check.
-        var tmpName = subscriptionName.Replace(@"\", pathDelimiter);
-        if (tmpName.Length > ruleNameMaximumLength)
-            throw new ArgumentOutOfRangeException(subscriptionName,
-                $@"Subscribe name '{subscriptionName}' exceeds the '{ruleNameMaximumLength}' character limit.");
-
-        if (tmpName.StartsWith(pathDelimiter, StringComparison.OrdinalIgnoreCase) ||
-            tmpName.EndsWith(pathDelimiter, StringComparison.OrdinalIgnoreCase))
-            throw new ArgumentException(
-                $@"The subscribe name cannot contain '/' as prefix or suffix. The supplied value is '{subscriptionName}'",
-                subscriptionName);
-
-        if (tmpName.Contains(pathDelimiter))
-            throw new ArgumentException($@"The subscribe name contains an invalid character '{pathDelimiter}'",
-                subscriptionName);
-
-        foreach (var uriSchemeKey in invalidEntityPathCharacters)
-        {
-            if (subscriptionName.IndexOf(uriSchemeKey) >= 0)
-                throw new ArgumentException(
-                    $@"'{subscriptionName}' contains character '{uriSchemeKey}' which is not allowed because it is reserved in the Uri scheme.",
-                    subscriptionName);
-        }
-    }
-
     #endregion private methods
 }

src/DotNetCore.CAP.AzureServiceBus/CAP.AzureServiceBusOptions.cs

@@ -32,6 +32,14 @@ public class AzureServiceBusOptions
     /// </summary>
     public string Namespace { get; set; } = default!;
 
+    /// <summary>
+    /// When set to <c>true</c> (default), CAP will automatically create topics, subscriptions,
+    /// and rules using the <see cref="Azure.Messaging.ServiceBus.Administration.ServiceBusAdministrationClient"/>.
+    /// Set to <c>false</c> to skip automatic provisioning, which is useful when the admin API
+    /// is unavailable (e.g., Azure Service Bus Emulator).
+    /// </summary>
+    public bool AutoProvision { get; set; } = true;
+
     /// <summary>
     /// Whether Service Bus sessions are enabled. If enabled, all messages must contain a
     /// <see cref="AzureServiceBusHeaders.SessionId" /> header. Defaults to false.