Enhanced default resilient producer options

Author: smbecker

src/DotPulsar.Extensions.Resiliency/Extensions/PulsarResilientExtensions.cs

@@ -38,44 +38,40 @@ public static ValueTask Process<TMessage>(
 		CancellationToken cancellationToken = default)
 		=> Process(consumer, processor, resiliencePipeline, new ProcessingOptions(), failureHandler, cancellationToken);
 
-	public static ResiliencePipelineBuilder AddResilientProducer(this ResiliencePipelineBuilder pipelineBuilder, Action<RetryStrategyOptions>? configureRetry = null, Action<TimeoutStrategyOptions>? configureTimeout = null) {
-		return pipelineBuilder.AddProducerRetry(configureRetry).AddProducerTimeout(configureTimeout);
-	}
-
-	public static ResiliencePipelineBuilder AddProducerTimeout(this ResiliencePipelineBuilder pipelineBuilder, Action<TimeoutStrategyOptions>? configure = null) {
-		var options = new TimeoutStrategyOptions {
+	public static ResiliencePipelineBuilder AddResilientProducerDefaults(this ResiliencePipelineBuilder pipelineBuilder, Action<RetryStrategyOptions>? configureRetry = null, Action<TimeoutStrategyOptions>? configureTimeout = null) {
+		var timeoutOptions = new TimeoutStrategyOptions {
 			Timeout = TimeSpan.FromSeconds(30)
 		};
-		configure?.Invoke(options);
-		return pipelineBuilder.AddTimeout(options);
-	}
+		configureTimeout?.Invoke(timeoutOptions);
+		pipelineBuilder.AddTimeout(timeoutOptions);
 
-	public static ResiliencePipelineBuilder AddProducerRetry(this ResiliencePipelineBuilder pipelineBuilder, Action<RetryStrategyOptions>? configure = null) {
-		var options = new RetryStrategyOptions {
+		var retryOptions = new RetryStrategyOptions {
 			MaxRetryAttempts = 10,
 			Delay = TimeSpan.FromMilliseconds(100),
 			MaxDelay = TimeSpan.FromMilliseconds(5000),
 			BackoffType = DelayBackoffType.Exponential,
 			ShouldHandle = static args => {
 				var ex = args.Outcome.Exception;
-				if (ShouldHandleProducerException(ex)) {
+				if (ShouldRetryProducerException(ex)) {
 					return new ValueTask<bool>(true);
 				}
 
 				return new ValueTask<bool>(false);
 			}
 		};
-		configure?.Invoke(options);
-		return pipelineBuilder.AddRetry(options);
+		configureRetry?.Invoke(retryOptions);
+		pipelineBuilder.AddRetry(retryOptions);
+
+		return pipelineBuilder;
 	}
 
-	public static bool ShouldHandleProducerException(Exception? exception) {
+	public static bool ShouldRetryProducerException(Exception? exception) {
 		return exception is not ResilientProducerDisposedException
 			&& exception is not PulsarClientDisposedException
 			&& exception is ProducerFaultedException or ProducerClosedException or ObjectDisposedException;
 	}
 
-	public static IProducer<TMessage> CreateResilient<TMessage>(this IProducerBuilder<TMessage> producerBuilder, ResiliencePipeline? resiliencePipeline = null) {
+	public static IProducer<TMessage> CreateResilient<TMessage>(this IProducerBuilder<TMessage> producerBuilder, ResiliencePipeline? resiliencePipeline) {
 		ArgumentNullException.ThrowIfNull(producerBuilder);
 
 		if (resiliencePipeline == null || Equals(resiliencePipeline, ResiliencePipeline.Empty)) {
@@ -89,7 +85,7 @@ public static IProducer<TMessage> CreateResilient<TMessage>(this IProducerBuilde
 		if (configurePipelineBuilder != null) {
 			configurePipelineBuilder(pipelineBuilder);
 		} else {
-			pipelineBuilder.AddResilientProducer();
+			pipelineBuilder.AddResilientProducerDefaults();
 		}
 
 		return CreateResilient(producerBuilder, pipelineBuilder.Build());

src/DotPulsar.Extensions.Resiliency/Internal/ResilientProducer.cs

@@ -140,11 +140,12 @@ public ResilienceSendChannel(ResilientProducer<TMessage> producer) {
 			this.producer = producer;
 		}
 
-		public async ValueTask Send(MessageMetadata metadata, TMessage message, Func<MessageId, ValueTask>? onMessageSent, CancellationToken cancellationToken) {
-			var messageId = await producer.Send(metadata, message, cancellationToken).ConfigureAwait(false);
-			if (onMessageSent != null) {
-				await onMessageSent(messageId).ConfigureAwait(false);
-			}
+		public ValueTask Send(MessageMetadata metadata, TMessage message, Func<MessageId, ValueTask>? onMessageSent, CancellationToken cancellationToken) {
+			return producer.resiliencePipeline.ExecuteAsync(static (state, ct) => {
+				var (sendChannel, message, metadata, onMessageSent) = state;
+				var producer = sendChannel.producer.GetOrCreateProducer();
+				return producer.SendChannel.Send(metadata, message, onMessageSent, ct);
+			}, (this, message, metadata, onMessageSent), cancellationToken);
 		}
 
 		public void Complete() {

src/DotPulsar.Extensions.Resiliency/README.md

@@ -21,14 +21,14 @@ await using var dlq = new DeadLetterPolicy(
 ...
 await dlq.ReconsumeLater(message);
 ```
-You will need to have a separate consumer set up to listen to the retry topic in order to process the retry messages. If those messages fail, then you will need to re-submit them to the DLQ to mark them as retry or dead.
+You will need to have a separate consumer set up to listen to the retry topic in order to process the retry messages. If those messages fail, then you will need to re-submit them to the `DeadLetterPolicy` to mark them as retry or dead.
 
 **Using [Polly](https://www.pollydocs.org/) to create resilient producers:**
 ```c#
 await using var producer = client.NewProducer(Schema.String)
 	.Topic("...")
-	.CreateResilient(pipeline => {
-		pipeline.AddRetryProducer(options => {
+	.CreateResilient(static pipeline => {
+		pipeline.AddResilientProducerDefaults(configureRetry: static options => {
 			options.MaxRetryAttempts = 3;
 		});
 	});