Please support Polly

[mviegas]

Hey @ivanpaulovich was looking for an issue to solve and participate, and this one just caught my attention. Could you describe it with more details?!

[ivanpaulovich]

Hi @mviegas :)

This task would require some prototyping and possible redesign. I would like to add a retry mechanism to the publish/send methods.

A simple code using Polly would be like:

using Polly;

// code omitted

//
// Setup FluentMediator

var services = new ServiceCollection();
services.AddFluentMediator(builder =>
{
    builder.On<PingRequest>().PipelineAsync()
        .Call<IPingHandler>(async (handler, req) => await handler.MyCustomFooBarAsync(req))
        .Build();
});
var pingHandler = new Mock<IPingHandler>();
services.AddScoped(provider => pingHandler.Object);

var provider = services.BuildServiceProvider();
var mediator = provider.GetRequiredService<IMediator>();

var ping = new PingRequest("Async Ping");

//
// Setup Polly

var retryPolicy = Policy
    .HandleAsync<Exception>() 
    .Retry(3);

var fallbackPolicy = Policy
    .HandleAsync<Exception>()
    .Fallback((cancellationToken) => Console.WriteLine("An error happened"));

//
// Invoke Mediator within a Polly Policy
// In case of exceptions it would be retried 3 times then call console.Write line

await fallbackPolicy
    .Wrap(retryPolicy)
    .ExecuteAsync(async () => await mediator.PublishAsync(ping)); 

The previous code needs some testing, and as you seem it became very verbose to invoke the PublishAsync.

I wish we could have an AddPolly method to the pipeline so we don't need this when invoking the Publish/Send.

[ivanpaulovich]

I think the first step is to set up SimpleConsoleApp with https://github.com/App-vNext/Polly library then see what we can do on our library to make it easier

[mviegas]

Oh, now I got your point. I'll work on it within the next weekend and try to update something here!

[mviegas]

Well, it's been a long time hahaha but talking about time, now is when I finally found some to think about it. So today i've been playing with some things like you suggested in your example and came with the following extension method:

public static IPipelineAsyncBuilder<TRequest> CallWithPolicyAsync<THandler, TRequest>(
    this IPipelineAsyncBuilder<TRequest> builder,
    Func<THandler, TRequest, Task> handler,
    Func<Task> fallbackAction = default)
{
    var retryPolicy = Policy.Handle<Exception>().RetryAsync(3);

    var fallbackPolicy = Policy.Handle<Exception>().FallbackAsync(async _ => await fallbackAction());

    var policyAction = new Func<THandler, TRequest, Task>(async (h, r) => await fallbackPolicy
            .WrapAsync(retryPolicy)
            .ExecuteAsync(async () =>
            {
                await handler.Invoke(h, r);
            }));

    builder.Call<THandler>(async (h, r) => await policyAction(h, r));

    return builder;
}

Then, on SimpleConsoleApp I just call:

builder.On<PingRequest>()
.PipelineAsync()
.CallWithPolicyAsync<PingHandler, PingRequest>(async (handler, req) => await handler.MyMethodAsync(req));

I think it I made it through a nice proof of concept to see if its possible to implement it through extension methods. Now I would like to ask for your opinion @ivanpaulovich on which things might be useful for a v1 PR. I thought about:

• Set Policy type and parameters depending on type (maybe create a PolicyBuilder that can be configured through some option chaining).
• I saw that we have Pipeline, PipelineAsync and CancellablePipelineAsync. Regarding this, I think that we should have one method extension method to add policies to each one of these builders, right? Or do you see a way to share one single method between them?

Thanks for the help!

[mviegas]

Just went through some more brainstorming over here:

public static IPipelineBuilder<TRequest> CallWithPolicy<THandler, TRequest>(
    this IPipelineBuilder<TRequest> builder,
    Action<THandler, TRequest> handler,
    Func<THandler, TRequest, Policy> mainPolicyConfiguration,
    params Policy[] policiesToWrap)
{
    if (builder is null) throw new ArgumentNullException(nameof(builder));

    if (mainPolicyConfiguration is null) throw new ArgumentNullException(nameof(mainPolicyConfiguration));

    var configuredPolicy = new Action<THandler, TRequest>((h, r) =>
    {
        var policy = mainPolicyConfiguration.Invoke(h, r);

        foreach (var policyToWrap in policiesToWrap)
        {
            policy.Wrap(policyToWrap);
        }

        policy.Execute(() => handler(h, r));
    });

    builder.Call<THandler>((h, r) => configuredPolicy(h, r));

    return builder;
}

On setup:

builder
    .On<PingRequest>()
    .Pipeline()
    .CallWithPolicy<PingHandler, PingRequest>(
        (handler, req) => handler.MyMethod(req),
        (_, req) => Policy.Handle<Exception>().Fallback(() => Console.WriteLine($"This is a fallback for attempt #{req.Count}")));

I have to confess that I have some ambiguous feelings about this approach:

• It is flexible. One can setup policies and their executions at its own taste.
• It is very verbose.

[ivanpaulovich]

Hi @mviegas,

How was the vacation? I hope you had a good time 🍺 This is an interesting implementation and it does not bring dependencies to the Core FluentMediator.

I guess the next step is to create a FluentMediator.Polly project so we design the nuget package and try out the implementation? Would you open a PR?

[mviegas]

Hey @ivanpaulovich thanks for the reply! The vacation time was great, need to destress after the times we had over here due to quarantine.

I'll prepare this project along the week and submit a PR so we can try this out better!