Merge pull request #589 from Particular/many-transports-fix-v3.1

Fix address mappings for when more than two transports are being used - 3.1

Author: jpalac

.github/workflows/ci.yml

@@ -89,7 +89,7 @@ jobs:
           echo "AWS_REGION=${{ secrets.AWS_REGION }}" | Out-File -FilePath $Env:GITHUB_ENV -Encoding utf-8 -Append
       - name: Setup RabbitMQ
         if: matrix.transport == 'RabbitMQ'
-        uses: Particular/setup-rabbitmq-action@v1.7.0
+        uses: Particular/setup-rabbitmq-action@v1.7.1
         with:
           connection-string-name: RabbitMQTransport_ConnectionString
           tag: RabbitMQTransportBridge

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/AcceptanceTests/Shared/Support/BridgeAcceptanceTest.cs

@@ -41,7 +41,6 @@ public Task TearDown()
     }
 
     protected TransportDefinition TransportBeingTested => bridgeTransportDefinition.TransportDefinition;
-    protected TransportDefinition TestTransport;
 
     BridgeTransportDefinition bridgeTransportDefinition;
 }

src/AcceptanceTests/Shared/Support/DefaultTestServer.cs

@@ -8,6 +8,8 @@
 
 public class DefaultTestServer : IEndpointSetupTemplate
 {
+    const string StorageDirectory = "DefaultTestingTransport";
+
 #pragma warning disable PS0013 // Add a CancellationToken parameter type argument
     public virtual async Task<EndpointConfiguration> GetConfiguration(RunDescriptor runDescriptor, EndpointCustomizationConfiguration endpointConfiguration, Func<EndpointConfiguration, Task> configurationBuilderCustomization)
 #pragma warning restore PS0013 // Add a CancellationToken parameter type argument
@@ -25,7 +27,7 @@ public virtual async Task<EndpointConfiguration> GetConfiguration(RunDescriptor
 
         configuration.RegisterComponentsAndInheritanceHierarchy(runDescriptor);
 
-        var transportDefinition = GetTestTransportDefinition();
+        var transportDefinition = GetTransportDefinition();
         configuration.UseTransport(transportDefinition);
 
         runDescriptor.OnTestCompleted(_ =>
@@ -43,12 +45,17 @@ public virtual async Task<EndpointConfiguration> GetConfiguration(RunDescriptor
         return configuration;
     }
 
-    public static AcceptanceTestingTransport GetTestTransportDefinition()
+    protected static AcceptanceTestingTransport GetTransportDefinition(string instanceId)
     {
         var testRunId = TestContext.CurrentContext.Test.ID;
+
         //make sure to run in a non-default directory to not clash with learning transport and other acceptance tests
-        var storageDir = Path.Combine(Path.GetTempPath(), testRunId, "DefaultTestingTransport");
+        var storagePath = Path.Combine(Path.GetTempPath(), testRunId, instanceId);
 
-        return new AcceptanceTestingTransport { StorageLocation = storageDir };
+        return new AcceptanceTestingTransport { StorageLocation = storagePath };
     }
+
+    protected virtual AcceptanceTestingTransport GetTransportDefinition() => GetTransportDefinition(StorageDirectory);
+
+    public static AcceptanceTestingTransport GetTestTransportDefinition() => GetTransportDefinition(StorageDirectory);
 }

src/AcceptanceTests/Shared/Support/ReceivingTestServer.cs

@@ -0,0 +1,10 @@
+using NServiceBus;
+
+public class ReceivingTestServer : DefaultTestServer
+{
+    const string StorageDirectory = "ReceiverTestingTransport";
+
+    protected override AcceptanceTestingTransport GetTransportDefinition() => GetTransportDefinition(StorageDirectory);
+
+    public static AcceptanceTestingTransport GetReceivingTransportDefinition() => GetTransportDefinition(StorageDirectory);
+}

src/AcceptanceTests/Shared/Support/SendingTestServer.cs

@@ -0,0 +1,10 @@
+using NServiceBus;
+
+public class SendingTestServer : DefaultTestServer
+{
+    const string StorageDirectory = "SenderTestingTransport";
+
+    protected override AcceptanceTestingTransport GetTransportDefinition() => GetTransportDefinition(StorageDirectory);
+
+    public static AcceptanceTestingTransport GetSendingTransportDefinition() => GetTransportDefinition(StorageDirectory);
+}

src/AcceptanceTests/Shared/ThreeTransports.cs

@@ -0,0 +1,96 @@
+using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.AcceptanceTesting;
+using NUnit.Framework;
+using Conventions = NServiceBus.AcceptanceTesting.Customization.Conventions;
+
+public class ThreeTransports : BridgeAcceptanceTest
+{
+    [Test(Description = "Replicates issue reported in https://github.com/Particular/NServiceBus.MessagingBridge/issues/369")]
+    public async Task Should_translate_address_correctly_for_target_transport()
+    {
+        var endpointOnTestingTransportName = Conventions.EndpointNamingConvention(typeof(EndpointOnTestingTransport));
+        var endpointOnTransportUnderTestName = Conventions.EndpointNamingConvention(typeof(EndpointOnTransportUnderTest));
+
+        var options = new SendOptions();
+        options.SetDestination(Conventions.EndpointNamingConvention(typeof(ReceivingEndpoint)));
+
+        var ctx = await Scenario.Define<Context>()
+            .WithEndpoint<ReceivingEndpoint>()
+            .WithEndpoint<EndpointOnTestingTransport>(builder => builder
+                .When(c => c.EndpointsStarted, (session, _) => session.Send(new SomeMessage { From = endpointOnTestingTransportName }, options)))
+            .WithEndpoint<EndpointOnTransportUnderTest>(builder => builder
+                .When(c => c.EndpointsStarted, (session, _) => session.Send(new SomeMessage { From = endpointOnTransportUnderTestName }, options)))
+            .WithBridge(bridgeConfiguration =>
+            {
+                bridgeConfiguration.TranslateReplyToAddressForFailedMessages();
+
+                var receivingTransport = new TestableBridgeTransport(ReceivingTestServer.GetReceivingTransportDefinition())
+                {
+                    Name = "ReceivingTransport"
+                };
+                receivingTransport.AddTestEndpoint<ReceivingEndpoint>();
+                bridgeConfiguration.AddTransport(receivingTransport);
+
+                var acceptanceTestingTransport = new TestableBridgeTransport(SendingTestServer.GetSendingTransportDefinition())
+                {
+                    Name = "SendingAcceptanceTestingTransportName"
+                };
+                acceptanceTestingTransport.AddTestEndpoint<EndpointOnTestingTransport>();
+                bridgeConfiguration.AddTransport(acceptanceTestingTransport);
+
+                var transportUnderTest = new TestableBridgeTransport(TransportBeingTested)
+                {
+                    Name = "TransportUnderTest"
+                };
+                transportUnderTest.AddTestEndpoint<EndpointOnTransportUnderTest>();
+                bridgeConfiguration.AddTransport(transportUnderTest);
+            })
+            .Done(c => c.ReceivedMessageCount == 2)
+            .Run();
+
+    }
+
+    public class ReceivingEndpoint : EndpointConfigurationBuilder
+    {
+        public ReceivingEndpoint() => EndpointSetup<ReceivingTestServer>();
+
+        class SomeMessageHandler(Context context) : IHandleMessages<SomeMessage>
+        {
+            public Task Handle(SomeMessage message, IMessageHandlerContext context)
+            {
+                Assert.Multiple(() =>
+                {
+                    Assert.That(context.MessageHeaders.TryGetValue("NServiceBus.ReplyToAddress", out var headerValue), Is.True);
+                    Assert.That(headerValue, Is.EqualTo(message.From));
+                });
+
+                testContext.ReceivedMessageCount++;
+
+                return Task.CompletedTask;
+            }
+
+            readonly Context testContext = context;
+        }
+    }
+
+    public class EndpointOnTestingTransport : EndpointConfigurationBuilder
+    {
+        public EndpointOnTestingTransport() => EndpointSetup<SendingTestServer>();
+    }
+
+    public class EndpointOnTransportUnderTest : EndpointConfigurationBuilder
+    {
+        public EndpointOnTransportUnderTest() => EndpointSetup<DefaultServer>();
+    }
+
+    public class Context : ScenarioContext
+    {
+        public int ReceivedMessageCount { get; set; }
+    }
+
+    public class SomeMessage : IMessage
+    {
+        public string From { get; set; }
+    }
+}

src/NServiceBus.MessagingBridge/AddressMap.cs

@@ -0,0 +1,61 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using NServiceBus;
+using NServiceBus.Raw;
+using NServiceBus.Transport;
+
+/// <summary>
+/// How to map an address depends upon which transport the message is being sent to, 
+/// so it is necessary to have an address mapping dictionary for each transport
+/// </summary>
+class AddressMap(IReadOnlyDictionary<string, IStartableRawEndpoint> dispatchers) : IAddressMap
+{
+    readonly Dictionary<string, TransportAddressMap> addressMap =
+        dispatchers.ToDictionary(dispatcher => dispatcher.Key, _ => new TransportAddressMap());
+
+    readonly HashSet<string> endpoints = [];
+
+    public void Add(BridgeTransport transport, BridgeEndpoint endpoint)
+    {
+        if (!endpoints.Add(endpoint.Name))
+        {
+            throw new InvalidOperationException($"{endpoint.Name} has already been added to the address map.");
+        }
+
+        var queueAddress = new QueueAddress(endpoint.Name);
+
+        foreach (var targetTransport in addressMap.Keys)
+        {
+            var targetAddress = DetermineTransportAddress(targetTransport, transport.Name, endpoint.QueueAddress, queueAddress);
+
+            foreach (var sourceTransport in addressMap.Keys)
+            {
+                var sourceAddress = DetermineTransportAddress(sourceTransport, transport.Name, endpoint.QueueAddress, queueAddress);
+
+                addressMap[targetTransport][sourceAddress] = targetAddress;
+            }
+        }
+    }
+
+    string DetermineTransportAddress(string transportToMap, string endpointTransport, string overriddenQueueAddress, QueueAddress queueAddress) => transportToMap == endpointTransport
+        ? overriddenQueueAddress ?? dispatchers[transportToMap].ToTransportAddress(queueAddress)
+        : dispatchers[transportToMap].ToTransportAddress(queueAddress);
+
+    public bool TryTranslate(string targetTransport, string address, out string bestMatch)
+    {
+        var transportAddressMappings = addressMap[targetTransport];
+
+        if (transportAddressMappings.TryGetValue(address, out bestMatch))
+        {
+            return true;
+        }
+
+        bestMatch = address.GetClosestMatch(transportAddressMappings.Keys);
+        return false;
+    }
+}
+
+class TransportAddressMap : Dictionary<string, string>
+{
+}

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);