Add ability to specify a hierarchy namespace

Author: abparticular

src/Transport/AzureServiceBusTransport.cs

@@ -78,7 +78,9 @@ public override async Task<TransportInfrastructure> Initialize(HostSettings host
             {
                 TransportType = transportType,
                 EnableCrossEntityTransactions = enableCrossEntityTransactions,
-                Identifier = $"Client-{receiver.Id}-{receiver.ReceiveAddress}-{Guid.NewGuid()}",
+                Identifier = HierarchyNamespace is null
+                    ? $"Client-{receiver.Id}-{receiver.ReceiveAddress}-{Guid.NewGuid()}"
+                    : $"Client-{HierarchyNamespace}-{receiver.Id}-{receiver.ReceiveAddress}-{Guid.NewGuid()}",
             };
             ApplyRetryPolicyOptionsIfNeeded(receiveClientOptions);
             ApplyWebProxyIfNeeded(receiveClientOptions);
@@ -93,7 +95,9 @@ public override async Task<TransportInfrastructure> Initialize(HostSettings host
             TransportType = transportType,
             // for the default client we never want things to automatically use cross entity transaction
             EnableCrossEntityTransactions = false,
-            Identifier = $"Client-{hostSettings.Name}-{Guid.NewGuid()}"
+            Identifier = HierarchyNamespace is null
+                ? $"Client-{hostSettings.Name}-{Guid.NewGuid()}"
+                : $"Client-{HierarchyNamespace}-{hostSettings.Name}-{Guid.NewGuid()}"
         };
         ApplyRetryPolicyOptionsIfNeeded(defaultClientOptions);
         ApplyWebProxyIfNeeded(defaultClientOptions);
@@ -105,7 +109,7 @@ public override async Task<TransportInfrastructure> Initialize(HostSettings host
             ? new ServiceBusAdministrationClient(FullyQualifiedNamespace, TokenCredential)
             : new ServiceBusAdministrationClient(ConnectionString);
 
-        var infrastructure = new AzureServiceBusTransportInfrastructure(this, hostSettings, receiveSettingsAndClientPairs, defaultClient, administrationClient);
+        var infrastructure = new AzureServiceBusTransportInfrastructure(this, hostSettings, receiveSettingsAndClientPairs, defaultClient, administrationClient, HierarchyNamespace);
 
         if (hostSettings.SetupInfrastructure)
         {
@@ -114,7 +118,7 @@ await administrationClient.AssertNamespaceManageRightsAvailable(cancellationToke
 
             var allQueues = infrastructure.Receivers
                 .Select(r => r.Value.ReceiveAddress)
-                .Concat(sendingAddresses)
+                .Concat(sendingAddresses.Select(s => HierarchyNamespace is null || s.StartsWith(HierarchyNamespace) ? s : $"{HierarchyNamespace}/{s}"))
                 .ToArray();
 
             var queueCreator = new TopologyCreator(this);
@@ -193,6 +197,13 @@ internal set
 
     TopicTopology topology;
 
+    /// <summary>
+    /// Specifies the value that will be prepended to every entity name referenced by the endpoint.
+    /// This will ensure that all entities will be part of the <see href="https://learn.microsoft.com/en-us/rest/api/servicebus/addressing-and-protocol">same hierarchy</see> 
+    /// within the ServiceBus namespace.
+    /// </summary>
+    public string? HierarchyNamespace { get; set; }
+
     /// <summary>
     /// The maximum size used when creating queues and topics in GB.
     /// </summary>

src/Transport/AzureServiceBusTransportInfrastructure.cs

@@ -16,21 +16,24 @@ sealed class AzureServiceBusTransportInfrastructure : TransportInfrastructure
     readonly HostSettings hostSettings;
     readonly ServiceBusClient defaultClient;
     readonly ServiceBusAdministrationClient administrationClient;
+    readonly string? hierarchyNamespace;
     readonly (ReceiveSettings receiveSettings, ServiceBusClient client)[] receiveSettingsAndClientPairs;
 
     public AzureServiceBusTransportInfrastructure(
         AzureServiceBusTransport transportSettings,
         HostSettings hostSettings,
         (ReceiveSettings receiveSettings, ServiceBusClient client)[] receiveSettingsAndClientPairs,
         ServiceBusClient defaultClient,
-        ServiceBusAdministrationClient administrationClient
+        ServiceBusAdministrationClient administrationClient,
+        string? hierarchyNamespace
     )
     {
         this.transportSettings = transportSettings;
 
         this.hostSettings = hostSettings;
         this.defaultClient = defaultClient;
         this.administrationClient = administrationClient;
+        this.hierarchyNamespace = hierarchyNamespace;
         this.receiveSettingsAndClientPairs = receiveSettingsAndClientPairs;
 
         messageSenderRegistry = new MessageSenderRegistry();
@@ -39,6 +42,7 @@ ServiceBusAdministrationClient administrationClient
             defaultClient,
             messageSenderRegistry,
             transportSettings.Topology,
+            hierarchyNamespace,
             transportSettings.OutgoingNativeMessageCustomization
         );
         Receivers = receiveSettingsAndClientPairs.ToDictionary(static settingsAndClient =>
@@ -138,7 +142,9 @@ await Task.WhenAll(Receivers.Values.Select(r => r.StopReceive(cancellationToken)
 
     public override string ToTransportAddress(QueueAddress address)
     {
-        var queue = new StringBuilder(address.BaseAddress);
+        var queue = hierarchyNamespace is null
+            ? new StringBuilder(address.BaseAddress)
+            : new StringBuilder($"{hierarchyNamespace}/{address.BaseAddress}");
 
         if (address.Discriminator != null)
         {

src/Transport/Sending/MessageDispatcher.cs

@@ -13,6 +13,7 @@ class MessageDispatcher(
     ServiceBusClient defaultClient,
     MessageSenderRegistry messageSenderRegistry,
     TopicTopology topology,
+    string? hierarchyNamespace,
     OutgoingNativeMessageCustomizationAction? customizerCallback = null)
     : IMessageDispatcher
 {
@@ -31,9 +32,7 @@ public async Task Dispatch(TransportOperations outgoingMessages, TransportTransa
         var unicastTransportOperations = outgoingMessages.UnicastTransportOperations;
         var multicastTransportOperations = outgoingMessages.MulticastTransportOperations;
 
-        var transportOperations =
-            new List<IOutgoingTransportOperation>(unicastTransportOperations.Count +
-                                                  multicastTransportOperations.Count);
+        var transportOperations = new List<IOutgoingTransportOperation>(unicastTransportOperations.Count + multicastTransportOperations.Count);
         transportOperations.AddRange(unicastTransportOperations);
         transportOperations.AddRange(multicastTransportOperations);
 
@@ -44,7 +43,7 @@ public async Task Dispatch(TransportOperations outgoingMessages, TransportTransa
 
         foreach (var operation in transportOperations)
         {
-            var destination = operation.ExtractDestination(topology);
+            var destination = operation.ExtractDestination(topology, hierarchyNamespace);
             switch (operation.RequiredDispatchConsistency)
             {
                 case DispatchConsistency.Default:

src/Transport/Sending/OutgoingTransportOperationExtensions.cs

@@ -1,11 +1,14 @@
 namespace NServiceBus.Transport.AzureServiceBus;
 
 using System;
+using System.Collections.Concurrent;
 using Azure.Messaging.ServiceBus;
 using Logging;
 
 static class OutgoingTransportOperationExtensions
 {
+    static readonly ConcurrentDictionary<string, string> destinationCache = [];
+
     public static void ApplyCustomizationToOutgoingNativeMessage(
         this IOutgoingTransportOperation transportOperation,
         ServiceBusMessage message, TransportTransaction transportTransaction, ILog logger)
@@ -28,14 +31,19 @@ public static void ApplyCustomizationToOutgoingNativeMessage(
     }
 
     public static string ExtractDestination(this IOutgoingTransportOperation outgoingTransportOperation,
-        TopicTopology topology)
+        TopicTopology topology,
+        string? hierarchyNamespace)
     {
+        string destination;
+
         switch (outgoingTransportOperation)
         {
             case MulticastTransportOperation multicastTransportOperation:
-                return topology.GetPublishDestination(multicastTransportOperation.MessageType);
+                destination = topology.GetPublishDestination(multicastTransportOperation.MessageType);
+                break;
+
             case UnicastTransportOperation unicastTransportOperation:
-                var destination = unicastTransportOperation.Destination;
+                destination = unicastTransportOperation.Destination;
 
                 // Workaround for reply-to address set by ASB transport
                 var index = unicastTransportOperation.Destination.IndexOf('@');
@@ -45,9 +53,16 @@ public static string ExtractDestination(this IOutgoingTransportOperation outgoin
                     destination = destination[..index];
                 }
 
-                return destination;
+                break;
+
             default:
                 throw new ArgumentOutOfRangeException(nameof(outgoingTransportOperation));
         }
+
+        return destinationCache.GetOrAdd(
+            destination,
+            static (dest, ns) => ns is null || dest.StartsWith(ns)
+                ? dest
+                : $"{ns}/{dest}", hierarchyNamespace);
     }
 }