Merge pull request #777 from Particular/forward-DiscardIfNotReceivedBefore

Forward DiscardIfNotReceivedBefore with dispatch

Author: WilliamBZA

src/AcceptanceTests/Shared/DoNotDeliverBefore.cs

@@ -0,0 +1,93 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.AcceptanceTesting;
+using NServiceBus.Features;
+using NUnit.Framework;
+using Conventions = NServiceBus.AcceptanceTesting.Customization.Conventions;
+
+public class DoNotDeliverBefore : BridgeAcceptanceTest
+{
+    static string OriginalEndpointName = Conventions.EndpointNamingConvention(typeof(OriginalEndpoint));
+
+    [Test]
+    public async Task Should_set_TTBR_correctly()
+    {
+        var ctx = await Scenario.Define<Context>()
+            .WithBridge((bridgeConfiguration, transportBeingTested) =>
+            {
+                transportBeingTested.HasEndpoint("_");
+                bridgeConfiguration.AddTransport(transportBeingTested);
+
+                bridgeConfiguration.AddTestTransportEndpoint(new BridgeEndpoint(OriginalEndpointName));
+            })
+            .WithEndpoint<MigratedEndpoint>()
+            .WithEndpoint<OriginalEndpoint>(endpoint => endpoint.When((session, context) =>
+            {
+                context.SendStartTime = DateTimeOffset.UtcNow;
+                return Task.CompletedTask;
+            }))
+            .Done(c => c.SendStartTime != DateTimeOffset.MinValue && DateTimeOffset.UtcNow >= c.SendStartTime.AddSeconds(10))
+            .Run();
+
+        Assert.That(ctx.NumberOfMessagesReceived, Is.EqualTo(1));
+    }
+
+    public class Context : BridgeScenarioContext
+    {
+        public int NumberOfMessagesReceived { get; set; }
+        public DateTimeOffset SendStartTime { get; set; }
+    }
+
+    public class MigratedEndpoint : EndpointConfigurationBuilder
+    {
+        public MigratedEndpoint()
+        {
+            CustomEndpointName(OriginalEndpointName);
+            EndpointSetup<DefaultTestServer>(c =>
+            {
+                // Set concurrency to 1 to ensure that the first message
+                // will delay sufficiently long for the TTBR setting
+                // to expire on the second message
+                c.LimitMessageProcessingConcurrencyTo(1);
+                c.PurgeOnStartup(true);
+            });
+        }
+
+        public class AMessageHandler(Context testContext) : IHandleMessages<AMessage>
+        {
+            public async Task Handle(AMessage message, IMessageHandlerContext context)
+            {
+                testContext.NumberOfMessagesReceived++;
+
+                await Task.Delay(5000);
+            }
+        }
+    }
+
+    public class OriginalEndpoint : EndpointConfigurationBuilder
+    {
+        public OriginalEndpoint() => EndpointSetup<DefaultServer>(cfg => cfg.EnableFeature<SendLocalFeature>());
+
+        public class SendLocalFeature : Feature
+        {
+            protected override void Setup(FeatureConfigurationContext context) => context.RegisterStartupTask(() => new SendLocalStartupTask());
+
+            class SendLocalStartupTask : FeatureStartupTask
+            {
+                protected override async Task OnStart(IMessageSession session, CancellationToken cancellationToken = default)
+                {
+                    await session.SendLocal(new AMessage(), cancellationToken);
+                    await session.SendLocal(new AMessage(), cancellationToken);
+                    await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
+                }
+
+                protected override Task OnStop(IMessageSession session, CancellationToken cancellationToken = default) => Task.CompletedTask;
+            }
+        }
+    }
+
+    [TimeToBeReceived("00:00:03")]
+    public class AMessage : IMessage;
+}

src/NServiceBus.MessagingBridge/TargetEndpointDispatcher.cs

@@ -1,5 +1,8 @@
-using System.Threading;
+using System;
+using System.Threading;
 using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.Performance.TimeToBeReceived;
 using NServiceBus.Raw;
 using NServiceBus.Routing;
 using NServiceBus.Transport;
@@ -22,6 +25,13 @@ public Task Dispatch(
         CancellationToken cancellationToken = default)
     {
         var transportOperation = new TransportOperation(outgoingMessage, targetAddress);
+
+        if (outgoingMessage.Headers.TryGetValue(Headers.TimeToBeReceived, out var timeToBeReceivedValue)
+            && TimeSpan.TryParse(timeToBeReceivedValue, out var timeToBeReceived))
+        {
+            transportOperation.Properties.DiscardIfNotReceivedBefore = new DiscardIfNotReceivedBefore(timeToBeReceived);
+        }
+
         return rawEndpoint.Dispatch(new TransportOperations(transportOperation), transaction, cancellationToken);
     }