Backport refactoring of bridge startup which includes creating dedicated dispatchers for each configured transport

Author: abparticular

src/AcceptanceTests/Shared/ReplyToAddress.cs

@@ -0,0 +1,94 @@
+using System;
+using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.AcceptanceTesting;
+using NServiceBus.Pipeline;
+using NUnit.Framework;
+using Conventions = NServiceBus.AcceptanceTesting.Customization.Conventions;
+
+public class ReplyToAddress : BridgeAcceptanceTest
+{
+    [Test]
+    public async Task Should_translate_address_for_already_migrated_endpoint()
+    {
+        var ctx = await Scenario.Define<Context>()
+            .WithEndpoint<SendingEndpoint>(builder =>
+            {
+                builder.DoNotFailOnErrorMessages();
+                builder.When(c => c.EndpointsStarted, (session, _) =>
+                {
+                    var options = new SendOptions();
+                    options.SetDestination(Conventions.EndpointNamingConvention(typeof(SecondMigratedEndpoint)));
+
+                    return session.Send(new ADelayedMessage(), options);
+                });
+            })
+            .WithEndpoint<FirstMigratedEndpoint>()
+            .WithEndpoint<SecondMigratedEndpoint>()
+            .WithBridge(bridgeConfiguration =>
+            {
+                var bridgeTransport = new TestableBridgeTransport(DefaultTestServer.GetTestTransportDefinition())
+                {
+                    Name = "DefaultTestingTransport"
+                };
+                bridgeTransport.AddTestEndpoint<SendingEndpoint>();
+                bridgeConfiguration.AddTransport(bridgeTransport);
+
+                var theOtherTransport = new TestableBridgeTransport(TransportBeingTested);
+                theOtherTransport.AddTestEndpoint<FirstMigratedEndpoint>();
+                theOtherTransport.AddTestEndpoint<SecondMigratedEndpoint>();
+                bridgeConfiguration.AddTransport(theOtherTransport);
+            })
+            .Done(c => c.ADelayedMessageReceived)
+            .Run();
+
+    }
+
+    public class SendingEndpoint : EndpointConfigurationBuilder
+    {
+        public SendingEndpoint() => EndpointSetup<DefaultTestServer>((c, runDescriptor) =>
+            c.Pipeline.Register(new OverrideReplyToAddress(), "Checks that the retry confirmation arrived"));
+
+        class OverrideReplyToAddress : Behavior<IOutgoingPhysicalMessageContext>
+        {
+            public override async Task Invoke(IOutgoingPhysicalMessageContext context, Func<Task> next)
+            {
+                context.Headers[Headers.ReplyToAddress] = Conventions.EndpointNamingConvention(typeof(FirstMigratedEndpoint));
+                await next();
+
+            }
+        }
+    }
+
+    public class FirstMigratedEndpoint : EndpointConfigurationBuilder
+    {
+        public FirstMigratedEndpoint() => EndpointSetup<DefaultServer>();
+    }
+
+    public class SecondMigratedEndpoint : EndpointConfigurationBuilder
+    {
+        public SecondMigratedEndpoint() => EndpointSetup<DefaultServer>();
+
+        class ADelayedMessageHandler : IHandleMessages<ADelayedMessage>
+        {
+            public ADelayedMessageHandler(Context context) => testContext = context;
+
+            public Task Handle(ADelayedMessage message, IMessageHandlerContext context)
+            {
+                testContext.ADelayedMessageReceived = true;
+                return Task.CompletedTask;
+            }
+
+            readonly Context testContext;
+        }
+    }
+
+    public class Context : ScenarioContext
+    {
+        public bool ADelayedMessageReceived { get; set; }
+    }
+
+    public class ADelayedMessage : IMessage
+    {
+    }
+}

src/NServiceBus.MessagingBridge/Configuration/BridgeConfiguration.cs

@@ -49,18 +49,14 @@ internal FinalizedBridgeConfiguration FinalizeConfiguration(ILogger<BridgeConfig
                 throw new InvalidOperationException("At least two transports needs to be configured");
             }
 
-            var tranportsWithNoEndpoints = transportConfigurations.Where(tc => !tc.Endpoints.Any())
-                .Select(t => t.Name).ToArray();
+            var allEndpoints = transportConfigurations
+                .SelectMany(t => t.Endpoints).ToArray();
 
-            if (tranportsWithNoEndpoints.Any())
+            if (!allEndpoints.Any())
             {
-                var endpointNames = string.Join(", ", tranportsWithNoEndpoints);
-                throw new InvalidOperationException($"At least one endpoint needs to be configured for transport(s): {endpointNames}");
+                throw new InvalidOperationException($"At least one endpoint needs to be configured");
             }
 
-            var allEndpoints = transportConfigurations
-                .SelectMany(t => t.Endpoints).ToArray();
-
             var duplicatedEndpoints = allEndpoints
                 .GroupBy(e => e.Name)
                 .Where(g => g.Count() > 1)

src/NServiceBus.MessagingBridge/EndpointProxyFactory.cs

@@ -58,6 +58,13 @@ public Task<IStartableRawEndpoint> CreateProxy(
         return RawEndpoint.Create(transportEndpointConfiguration, cancellationToken);
     }
 
+    public static Task<IStartableRawEndpoint> CreateDispatcher(BridgeTransport transportConfiguration, CancellationToken cancellationToken = default)
+    {
+        var endpointConfiguration = RawEndpointConfiguration.CreateSendOnly($"bridge-dispatcher-{transportConfiguration.Name}", transportConfiguration.TransportDefinition);
+
+        return RawEndpoint.Create(endpointConfiguration, cancellationToken);
+    }
+
     static bool IsSubscriptionMessage(IReadOnlyDictionary<string, string> messageContextHeaders)
     {
         var messageIntent = default(MessageIntent);

src/NServiceBus.MessagingBridge/EndpointRegistry.cs

@@ -1,55 +1,73 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
 using NServiceBus;
 using NServiceBus.Raw;
 using NServiceBus.Transport;
 
-class EndpointRegistry : IEndpointRegistry
+class EndpointRegistry(EndpointProxyFactory endpointProxyFactory, ILogger<StartableBridge> logger) : IEndpointRegistry
 {
-    public void RegisterDispatcher(
-        BridgeEndpoint endpoint,
-        string targetTransportName,
-        IStartableRawEndpoint rawEndpoint)
+    //NOTE: This method cannot have a return type of IAsyncEnumerable as all the endpoints need to be configured on the bridge before the bridge can start processing messages. 
+    public async Task<IEnumerable<ProxyRegistration>> Initialize(IReadOnlyCollection<BridgeTransport> transportConfigurations, CancellationToken cancellationToken = default)
     {
-        registrations.Add(new ProxyRegistration
-        {
-            Endpoint = endpoint,
-            TranportName = targetTransportName,
-            RawEndpoint = rawEndpoint
-        });
-    }
-
-    public void ApplyMappings(IReadOnlyCollection<BridgeTransport> transportConfigurations)
-    {
-        foreach (var registration in registrations)
-        {
-            var endpoint = registration.Endpoint;
-
-            // target transport is the transport where this endpoint is actually running
-            var targetTransport = transportConfigurations.Single(t => t.Endpoints.Any(e => e.Name == endpoint.Name));
+        // Assume that it is the number of endpoints that is more likely to scale up in size than the number of transports (typically only 2).
+        // Therefore in cases where it might matter, it is more efficient to iterate over the transports multiple times.
+        transportConfigurationMappings = transportConfigurations.ToDictionary(t => t.Name, t => t);
 
-            // just pick the first proxy that is running on the target transport since
-            // we just need to be able to send messages to that transport
-            var proxyEndpoint = registrations
-                .First(r => r.TranportName == targetTransport.Name)
-                .RawEndpoint;
+        IList<ProxyRegistration> proxyRegistrations = [];
 
-            //This value represents in fact the endpoint's name. It is wrapped in the QueueAddress class only because
-            //the ToTransportAddress API expects it.
-            var endpointName = new QueueAddress(endpoint.Name);
-
-            var transportAddress = endpoint.QueueAddress ?? proxyEndpoint.ToTransportAddress(endpointName);
-
-            endpointAddressMappings[registration.Endpoint.Name] = transportAddress;
-
-            targetEndpointAddressMappings[transportAddress] = registration.RawEndpoint.ToTransportAddress(endpointName);
-
-            targetEndpointDispatchers[registration.Endpoint.Name] = new TargetEndpointDispatcher(
-                targetTransport.Name,
-                proxyEndpoint,
-                transportAddress);
+        foreach (var targetTransport in transportConfigurations)
+        {
+            // Create the dispatcher for this transport
+            var dispatchEndpoint = await EndpointProxyFactory.CreateDispatcher(targetTransport, cancellationToken).ConfigureAwait(false);
+
+            proxyRegistrations.Add(new ProxyRegistration
+            {
+                Endpoint = null,
+                TranportName = targetTransport.Name,
+                RawEndpoint = dispatchEndpoint
+            });
+
+            // create required proxy endpoints on all transports
+            foreach (var endpointToSimulate in targetTransport.Endpoints)
+            {
+                // Endpoint will need to be proxied on the other transports
+                foreach (var proxyTransport in transportConfigurationMappings.Where(kvp => kvp.Key != targetTransport.Name).Select(kvp => kvp.Value))
+                {
+                    var startableEndpointProxy = await endpointProxyFactory.CreateProxy(endpointToSimulate, proxyTransport, cancellationToken)
+                        .ConfigureAwait(false);
+
+                    logger.LogInformation("Proxy for endpoint {endpoint} created on {transport}", endpointToSimulate.Name, proxyTransport.Name);
+
+                    var queueAddress = new QueueAddress(endpointToSimulate.Name);
+                    var targetTransportAddress = dispatchEndpoint.ToTransportAddress(queueAddress);
+                    var sourceTransportAddress = startableEndpointProxy.ToTransportAddress(queueAddress);
+
+                    endpointAddressMappings[endpointToSimulate.Name] = endpointToSimulate.QueueAddress ?? targetTransportAddress;
+                    targetEndpointAddressMappings[targetTransportAddress] = sourceTransportAddress;
+                    if (targetTransportAddress != sourceTransportAddress)
+                    {
+                        // Also add the reverse mapping so that any messages that were in-flight before a bridge configuration change
+                        // can still have their address translated correctly.  This also allows for the case where duplicate logical endpoints
+                        // are running as a competing consumer with the bridge in canary/parallel deployment scenarios.
+                        targetEndpointAddressMappings[sourceTransportAddress] = targetTransportAddress;
+                    }
+                    targetEndpointDispatchers[endpointToSimulate.Name] = new TargetEndpointDispatcher(targetTransport.Name, dispatchEndpoint, targetTransportAddress);
+
+                    proxyRegistrations.Add(new ProxyRegistration
+                    {
+                        Endpoint = endpointToSimulate,
+                        TranportName = proxyTransport.Name,
+                        RawEndpoint = startableEndpointProxy
+                    });
+                }
+            }
         }
+
+        return proxyRegistrations;
     }
 
     public TargetEndpointDispatcher GetTargetEndpointDispatcher(string sourceEndpointName)
@@ -96,12 +114,11 @@ static string GetClosestMatchForExceptionMessage(string sourceEndpointName, IEnu
         return nearestMatch ?? "(No mappings registered)";
     }
 
-    public IEnumerable<ProxyRegistration> Registrations => registrations;
+    Dictionary<string, BridgeTransport> transportConfigurationMappings = [];
 
     readonly Dictionary<string, TargetEndpointDispatcher> targetEndpointDispatchers = [];
     readonly Dictionary<string, string> targetEndpointAddressMappings = [];
     readonly Dictionary<string, string> endpointAddressMappings = [];
-    readonly List<ProxyRegistration> registrations = [];
 
     public class ProxyRegistration
     {

src/NServiceBus.MessagingBridge/RawEndpoints/InitializableRawEndpoint.cs

@@ -1,6 +1,7 @@
 namespace NServiceBus.Raw
 {
     using NServiceBus.Transport;
+    using System;
     using System.Collections.Generic;
     using System.Threading;
     using System.Threading.Tasks;
@@ -20,13 +21,18 @@ public async Task<IStartableRawEndpoint> Initialize(CancellationToken cancellati
                 null); //null means "not hosted by core", transport SHOULD adjust accordingly to not assume things
 
             var usePubSub = rawEndpointConfiguration.TransportDefinition.SupportsPublishSubscribe && !rawEndpointConfiguration.PublishAndSubscribeDisabled;
-            var receivers = new[]{
-                new ReceiveSettings(
-                    rawEndpointConfiguration.EndpointName,
-                    new QueueAddress(rawEndpointConfiguration.EndpointName),
-                    usePubSub,
-                    false,
-                    rawEndpointConfiguration.PoisonMessageQueue)};
+            var receivers = Array.Empty<ReceiveSettings>();
+
+            if (!rawEndpointConfiguration.SendOnly)
+            {
+                receivers = [
+                    new ReceiveSettings(
+                        rawEndpointConfiguration.EndpointName,
+                        new QueueAddress(rawEndpointConfiguration.EndpointName),
+                        usePubSub,
+                        false,
+                        rawEndpointConfiguration.PoisonMessageQueue)];
+            }
 
             var sendingQueues = new List<string>(rawEndpointConfiguration.AdditionalQueues);
 

src/NServiceBus.MessagingBridge/RawEndpoints/StartableRawEndpoint.cs

@@ -14,12 +14,12 @@ public StartableRawEndpoint(
             RawCriticalError criticalError)
         {
             this.criticalError = criticalError;
-            messagePump = transportInfrastructure.Receivers.Values.First();
+            messagePump = transportInfrastructure.Receivers.Values.FirstOrDefault();
             dispatcher = transportInfrastructure.Dispatcher;
             this.rawEndpointConfiguration = rawEndpointConfiguration;
             this.transportInfrastructure = transportInfrastructure;
-            SubscriptionManager = messagePump.Subscriptions;
-            TransportAddress = messagePump.ReceiveAddress;
+            SubscriptionManager = messagePump?.Subscriptions;
+            TransportAddress = messagePump?.ReceiveAddress;
         }
 
         public async Task<IReceivingRawEndpoint> Start(CancellationToken cancellationToken = default)