diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresQueueDescriptorConfiguration.cs b/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresQueueDescriptorConfiguration.cs index 5d93a7cab6b..5876d07b152 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresQueueDescriptorConfiguration.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Configurations/PostgresQueueDescriptorConfiguration.cs @@ -67,6 +67,11 @@ internal PostgresQueueDescriptorConfiguration(string name) /// public List>> ReceivePipelineModifiers { get; } = []; + /// + /// Gets the configuration callbacks applied to this queue's receive endpoint, in registration order. + /// + public List> ConfigureReceiveEndpoint { get; } = []; + /// /// Gets the source topic bindings declared from this queue descriptor. /// diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresQueueDescriptor.cs b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresQueueDescriptor.cs index 22ec34ce547..0c88c5bcd3b 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresQueueDescriptor.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/IPostgresQueueDescriptor.cs @@ -2,8 +2,11 @@ namespace Mocha.Transport.Postgres; /// /// Fluent descriptor for configuring a PostgreSQL queue and its receive endpoint. +/// Receive configuration callbacks apply to the queue's receive endpoint. /// -public interface IPostgresQueueDescriptor : IMessagingDescriptor +public interface IPostgresQueueDescriptor + : IMessagingDescriptor + , IConfigurable { /// /// Sets whether the backing queue is automatically provisioned in the database. diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs index 7665c1aa08f..a2f5dd3bb8f 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresMessagingTransportDescriptor.cs @@ -340,6 +340,11 @@ private void ConfigureQueueEndpoint(PostgresQueueDescriptorConfiguration configu target.ReceivePipelineModifiers.AddRange(configuration.ReceivePipelineModifiers); CopyFaultEndpointFeature(configuration, target); CopySkippedEndpointFeature(configuration, target); + + foreach (var configure in configuration.ConfigureReceiveEndpoint) + { + configure(endpoint); + } } private static void CopyFaultEndpointFeature( diff --git a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs index 36ddcf89ed5..0090c05cbd0 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/Descriptors/PostgresQueueDescriptor.cs @@ -145,6 +145,14 @@ public IPostgresQueueDescriptor UseReceive( return this; } + /// + public void Configure(Action configure) + { + ArgumentNullException.ThrowIfNull(configure); + + Configuration.ConfigureReceiveEndpoint.Add(configure); + } + /// public IPostgresQueueDescriptor FaultEndpoint(Uri address) { diff --git a/src/Mocha/test/Mocha.Transport.Postgres.Tests/Descriptors/PostgresUnifiedQueueTests.cs b/src/Mocha/test/Mocha.Transport.Postgres.Tests/Descriptors/PostgresUnifiedQueueTests.cs index 6670cd5a389..6a85bb854dd 100644 --- a/src/Mocha/test/Mocha.Transport.Postgres.Tests/Descriptors/PostgresUnifiedQueueTests.cs +++ b/src/Mocha/test/Mocha.Transport.Postgres.Tests/Descriptors/PostgresUnifiedQueueTests.cs @@ -205,6 +205,66 @@ public void Queue_Should_PropagateAutoProvision_When_AutoProvisionCalled() Assert.Equal(false, queue.AutoProvision); } + [Fact] + public void Queue_Should_ConfigureEndpointReceiveFeatures_When_ResilienceConfiguredOnQueue() + { + // arrange + var runtime = CreateRuntime( + b => b.AddConsumer(), + t => + { + t.BindExplicitly(); + t.Queue("orders") + .Consumer() + .AddResilience(p => p.On().Discard()) + .AddCircuitBreaker(o => o.FailureRatio = 0.5) + .AddConcurrencyLimiter(o => o.MaxConcurrency = 4); + }); + var transport = runtime.Transports.OfType().Single(); + + // act + var endpoint = transport.ReceiveEndpoints + .OfType() + .Single(e => e.Queue.Name == "orders"); + + // assert + Type[] expectedRules = [typeof(Exception), typeof(InvalidOperationException)]; + Assert.Equal( + expectedRules, + endpoint.Features.Get()?.Rules.Select(r => r.ExceptionType)); + Assert.Equal(0.5, endpoint.Features.Get()?.FailureRatio); + Assert.Equal(4, endpoint.Features.Get()?.MaxConcurrency); + } + + [Fact] + public void Queue_Should_CombineExceptionPolicies_When_EndpointWithSameNameConfiguresResilience() + { + // arrange + var runtime = CreateRuntime( + b => b.AddConsumer(), + t => + { + t.BindExplicitly(); + t.Queue("orders") + .Consumer() + .AddResilience(p => p.On().Discard()); + t.Endpoint("orders").AddResilience(p => p.On().Discard()); + }); + var transport = runtime.Transports.OfType().Single(); + + // act + var endpoint = transport.ReceiveEndpoints + .OfType() + .Single(e => e.Queue.Name == "orders"); + + // assert + // endpoint callbacks run first, queue callbacks replay onto the same feature afterwards + Type[] expectedRules = [typeof(TimeoutException), typeof(Exception), typeof(InvalidOperationException)]; + Assert.Equal( + expectedRules, + endpoint.Features.Get()?.Rules.Select(r => r.ExceptionType)); + } + private static MessagingRuntime CreateRuntime( Action configureBuilder, Action configureTransport)