add filter

src/ProjectOrigin.Vault/Extensions/RandomDelayConfigurationExtensions.cs

@@ -0,0 +1,65 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using MassTransit;
+using MassTransit.Configuration;
+
+namespace ProjectOrigin.Vault.Extensions;
+
+public static class RandomDelayConfigurationExtensions
+{
+    public static void UseRandomDelay<TConsumer, TMessage>(this IConsumerConfigurator<TConsumer> configurator, int minDelayInMilliseconds = 1,
+        int maxDelayInMilliseconds = 1500)
+        where TConsumer : class, IConsumer<TMessage>
+        where TMessage : class
+    {
+        configurator.AddPipeSpecification(new RandomDelaySpecification<TConsumer, TMessage>(minDelayInMilliseconds, maxDelayInMilliseconds));
+    }
+}
+
+public class RandomDelaySpecification<TConsumer, TMessage>(int minDelayInMilliseconds, int maxDelayInMilliseconds)
+    : IPipeSpecification<ConsumerConsumeContext<TConsumer>>
+    where TConsumer : class, IConsumer<TMessage>
+    where TMessage : class
+{
+    public void Apply(IPipeBuilder<ConsumerConsumeContext<TConsumer>> builder)
+    {
+        builder.AddFilter(new RandomDelayFilter<TConsumer>(minDelayInMilliseconds, maxDelayInMilliseconds));
+    }
+
+    public IEnumerable<ValidationResult> Validate()
+    {
+        if (minDelayInMilliseconds < 0)
+            yield return this.Failure("RandomDelayFilter", "minDelayInMilliseconds cannot be negative");
+
+        if (maxDelayInMilliseconds <= minDelayInMilliseconds)
+            yield return this.Failure("RandomDelayFilter",
+                "maxDelayInMilliseconds must be greater than minDelayInMilliseconds");
+    }
+}
+
+public class RandomDelayFilter<TConsumer>(int minDelayInMilliseconds, int maxDelayInMilliseconds) : IFilter<ConsumerConsumeContext<TConsumer>>
+    where TConsumer : class
+{
+    public async Task Send(ConsumerConsumeContext<TConsumer> context,
+        IPipe<ConsumerConsumeContext<TConsumer>> next)
+    {
+        var delay = Random.Shared.Next(minDelayInMilliseconds, maxDelayInMilliseconds + 1);
+
+        context.GetOrAddPayload(() => new RandomDelayPayload { Delay = delay });
+
+        await Task.Delay(delay);
+        await next.Send(context);
+    }
+
+    public void Probe(ProbeContext context)
+    {
+        var scope = context.CreateFilterScope("random-delay");
+        scope.Add("description", $"Adds a random delay between {minDelayInMilliseconds}-{maxDelayInMilliseconds} ms to each message.");
+    }
+}
+
+public class RandomDelayPayload
+{
+    public int Delay { get; set; }
+}

src/ProjectOrigin.Vault/Startup.cs

@@ -113,6 +113,7 @@
             {
                 cfg.UseMessageRetry(r => r.Interval(20, TimeSpan.FromSeconds(5))
                     .Handle<QuantityNotYetAvailableToReserveException>());
+                cfg.UseRandomDelay<ClaimCertificateCommandHandler, ClaimCertificateCommand>();
             });
 
             o.AddConsumer<CheckForWithdrawnCertificatesCommandHandler>();
@@ -174,7 +175,7 @@
         app.ConfigureSqlMappers();
     }
 
     private void PrintNetworkOptions(IApplicationBuilder app)
     {
         var logger = app.ApplicationServices.GetRequiredService<ILogger<Startup>>();
         var options = app.ApplicationServices.GetRequiredService<IOptions<NetworkOptions>>().Value;

test/ProjectOrigin.Vault.Tests/Extensions/RandomDelayConfigurationExtensionsTests.cs

@@ -0,0 +1,84 @@
+using System.Linq;
+using System.Threading.Tasks;
+using MassTransit;
+using MassTransit.Internals;
+using MassTransit.Testing;
+using Microsoft.Extensions.DependencyInjection;
+using ProjectOrigin.Vault.Extensions;
+using Xunit;
+
+namespace ProjectOrigin.Vault.Tests.Extensions;
+
+public class RandomDelayConfigurationExtensionsTests
+{
+    [Fact]
+    public async Task Should_Apply_RandomDelay_To_Each_Message()
+    {
+        const int minDelayInMilliseconds = 5;
+        const int maxDelayInMilliseconds = 1000;
+        await using var provider = new ServiceCollection()
+            .AddMassTransitTestHarness(cfg =>
+            {
+                cfg.AddConsumer<TestConsumer>(consumerCfg =>
+                {
+                    consumerCfg.UseRandomDelay<TestConsumer, TestMessage>(minDelayInMilliseconds, maxDelayInMilliseconds);
+                });
+
+                cfg.UsingInMemory((context, inMemoryQueue) => { inMemoryQueue.ConfigureEndpoints(context); });
+            })
+            .BuildServiceProvider(true);
+
+        var harness = provider.GetRequiredService<ITestHarness>();
+        await harness.Start();
+
+        try
+        {
+            for (var i = 0; i < 5; i++)
+            {
+                await harness.Bus.Publish(new TestMessage());
+            }
+
+            Assert.True(await harness.Consumed.Any<TestMessage>(),
+                "Expected the test harness to consume TestMessage, but none was found.");
+
+            var consumerHarness = harness.GetConsumerHarness<TestConsumer>();
+
+            var consumedContexts = await consumerHarness
+                .Consumed
+                .SelectAsync<TestMessage>()
+                .Take(5)
+                .ToListAsync();
+
+            Assert.Equal(5, consumedContexts.Count);
+
+            var delays = consumedContexts
+                .Select(ctx =>
+                    ctx.Context.TryGetPayload<RandomDelayPayload>(out var payload)
+                        ? payload.Delay
+                        : 0)
+                .ToList();
+
+            Assert.Equal(5, delays.Count);
+
+            Assert.All(delays, d => Assert.InRange(d, minDelayInMilliseconds, maxDelayInMilliseconds));
+
+            Assert.Equal(5, delays.Distinct().Count());
+        }
+        finally
+        {
+            await harness.Stop();
+        }
+    }
+}
+
+public record TestMessage;
+
+public class TestConsumer : IConsumer<TestMessage>
+{
+    public Task Consume(ConsumeContext<TestMessage> context)
+    {
+        return Task.CompletedTask;
+    }
+}
+
+

test/ProjectOrigin.Vault.Tests/Extensions/RandomDelaySpecificationTests.cs

@@ -0,0 +1,47 @@
+using System.Threading.Tasks;
+using MassTransit;
+using ProjectOrigin.Vault.Extensions;
+using Xunit;
+
+namespace ProjectOrigin.Vault.Tests.Extensions
+{
+    public class RandomDelaySpecificationTests
+    {
+        [Fact]
+        public void Validate_ShouldReturnErrorIfMinDelayIsNegative()
+        {
+            var specification = new RandomDelaySpecification<TestConsumer, TestMessage>(-1, 100);
+
+            var error = Assert.Single(specification.Validate());
+
+            Assert.Equal("RandomDelayFilter", error.Key);
+            Assert.Equal("minDelayInMilliseconds cannot be negative", error.Message);
+        }
+
+        [Fact]
+        public void Validate_ShouldReturnErrorIfMaxDelayIsLessThanOrEqualToMinDelay()
+        {
+            var specification = new RandomDelaySpecification<TestConsumer, TestMessage>(5, 5);
+
+            var error = Assert.Single(specification.Validate());
+
+            Assert.Equal("RandomDelayFilter", error.Key);
+            Assert.Equal("maxDelayInMilliseconds must be greater than minDelayInMilliseconds", error.Message);
+        }
+
+        [Fact]
+        public void Validate_ShouldReturnNoErrorsForValidRange()
+        {
+            var specification = new RandomDelaySpecification<TestConsumer, TestMessage>(0, 10);
+
+            Assert.Empty(specification.Validate());
+        }
+
+        private class TestMessage;
+
+        private class TestConsumer : IConsumer<TestMessage>
+        {
+            public Task Consume(ConsumeContext<TestMessage> context) => Task.CompletedTask;
+        }
+    }
+}