How to change MediatR publish strategy

[AliNazar1991]

how is it possible to configure another publishing strategy when using MediatR? basically i need to configure it to use publishingStrategy.ParallelNoWait. any help would be appreciated since im not able to find any documentation regarding this.

[jbogard]

Are you using the DI extensions package? If so, you can specify the implementation type:

services.AddMediatR(opt => opt.Using<MyCustomMediatorType>(), typeof(Program));

[jbogard]

No the point of that sample was to highlight different possibilities of publish strategies and how they behave. Not to use that sample class as-is. You would create your own class, inherit from Mediator, override that PublishCore method and drop in whatever behavior you want.

[AliNazar1991]

ahh i see, thanks alot!
is it possible to configure a similar strategy, i.e fire and forget behaviour when using Send with requestHandlers instead of publish?

[jbogard]

No there's not for Send, by design. For the other strategies of "Send", like fire-and-forget, there are more appropriate solutions out there.

[AliNazar1991]

Currently im trying to achieve this fire-and-forget behaviour with Send by simply not awaiting the call, not sure if this is the right way to do it though..

[jbogard]

Yes that's the part of "more appropriate solutions" - don't use MediatR for fire-and-forget. You can use Task-related APIs for that instead, or even messaging/queueing. MediatR was never designed nor intended for that scenario.

[fiseni]

EDIT: Do not use this implementation, it has some flaws. You can find a better one here.

---

The example implementations are provided in the docs here. But, the example uses a separate construct Publisher. If you want to have it as an extension method to IMediatR, you can follow these steps:

• Add CustomMediator implementation (provided below).
• Register it in DI.

services.AddMediatR(options => options.Using<CustomMediator>().AsScoped(), typeof(Program));

• Add an extension method.

public static class MediatorExtensions
{
    public static Task Publish<TNotification>(this IMediator mediator, TNotification notification, PublishStrategy strategy, CancellationToken cancellationToken)
        where TNotification : INotification
    {
        if (mediator is CustomMediator customMediator)
        {
            return customMediator.Publish(notification, strategy, cancellationToken);
        }

        throw new NotImplementedException("The custom mediator implementation is not registered!");
    }
}

---

The CustomMediator implementation

public class CustomMediator : Mediator
{
    private readonly ServiceFactory _serviceFactory;
    private readonly Func<IEnumerable<Func<INotification, CancellationToken, Task>>, INotification, CancellationToken, Task> _publish;
    private readonly Dictionary<PublishStrategy, IMediator> _publishStrategies;

    private CustomMediator(ServiceFactory serviceFactory, Func<IEnumerable<Func<INotification, CancellationToken, Task>>, INotification, CancellationToken, Task> publish) : base(serviceFactory)
    {
        _serviceFactory = serviceFactory;
        _publish = publish;
        _publishStrategies = default!;
    }

    public CustomMediator(ServiceFactory serviceFactory) : base(serviceFactory)
    {
        _serviceFactory = serviceFactory;
        _publish = base.PublishCore;

        _publishStrategies = new();
        _publishStrategies[PublishStrategy.Async] = new CustomMediator(_serviceFactory, AsyncContinueOnException);
        _publishStrategies[PublishStrategy.ParallelNoWait] = new CustomMediator(_serviceFactory, ParallelNoWait);
        _publishStrategies[PublishStrategy.ParallelWhenAll] = new CustomMediator(_serviceFactory, ParallelWhenAll);
        _publishStrategies[PublishStrategy.ParallelWhenAny] = new CustomMediator(_serviceFactory, ParallelWhenAny);
        _publishStrategies[PublishStrategy.SyncContinueOnException] = new CustomMediator(_serviceFactory, SyncContinueOnException);
        _publishStrategies[PublishStrategy.SyncStopOnException] = new CustomMediator(_serviceFactory, SyncStopOnException);
    }

    protected override Task PublishCore(IEnumerable<Func<INotification, CancellationToken, Task>> allHandlers, INotification notification, CancellationToken cancellationToken)
    {
        return _publish(allHandlers, notification, cancellationToken);
    }

    public Task Publish<TNotification>(TNotification notification, PublishStrategy strategy, CancellationToken cancellationToken)
        where TNotification : INotification
    {
        if (!_publishStrategies.TryGetValue(strategy, out var mediator))
        {
            throw new ArgumentException($"Unknown strategy: {strategy}");
        }

        return mediator.Publish(notification, cancellationToken);
    }

    private Task ParallelWhenAll(IEnumerable<Func<INotification, CancellationToken, Task>> handlers, INotification notification, CancellationToken cancellationToken)
    {
        var tasks = new List<Task>();

        foreach (var handler in handlers)
        {
            tasks.Add(Task.Run(() => handler(notification, cancellationToken)));
        }

        return Task.WhenAll(tasks);
    }

    private Task ParallelWhenAny(IEnumerable<Func<INotification, CancellationToken, Task>> handlers, INotification notification, CancellationToken cancellationToken)
    {
        var tasks = new List<Task>();

        foreach (var handler in handlers)
        {
            tasks.Add(Task.Run(() => handler(notification, cancellationToken)));
        }

        return Task.WhenAny(tasks);
    }

    private Task ParallelNoWait(IEnumerable<Func<INotification, CancellationToken, Task>> handlers, INotification notification, CancellationToken cancellationToken)
    {
        foreach (var handler in handlers)
        {
            Task.Run(() => handler(notification, cancellationToken));
        }

        return Task.CompletedTask;
    }

    private async Task AsyncContinueOnException(IEnumerable<Func<INotification, CancellationToken, Task>> handlers, INotification notification, CancellationToken cancellationToken)
    {
        var tasks = new List<Task>();
        var exceptions = new List<Exception>();

        foreach (var handler in handlers)
        {
            try
            {
                tasks.Add(handler(notification, cancellationToken));
            }
            catch (Exception ex) when (!(ex is OutOfMemoryException || ex is StackOverflowException))
            {
                exceptions.Add(ex);
            }
        }

        try
        {
            await Task.WhenAll(tasks).ConfigureAwait(false);
        }
        catch (AggregateException ex)
        {
            exceptions.AddRange(ex.Flatten().InnerExceptions);
        }
        catch (Exception ex) when (!(ex is OutOfMemoryException || ex is StackOverflowException))
        {
            exceptions.Add(ex);
        }

        if (exceptions.Any())
        {
            throw new AggregateException(exceptions);
        }
    }

    private async Task SyncStopOnException(IEnumerable<Func<INotification, CancellationToken, Task>> handlers, INotification notification, CancellationToken cancellationToken)
    {
        foreach (var handler in handlers)
        {
            await handler(notification, cancellationToken).ConfigureAwait(false);
        }
    }

    private async Task SyncContinueOnException(IEnumerable<Func<INotification, CancellationToken, Task>> handlers, INotification notification, CancellationToken cancellationToken)
    {
        var exceptions = new List<Exception>();

        foreach (var handler in handlers)
        {
            try
            {
                await handler(notification, cancellationToken).ConfigureAwait(false);
            }
            catch (AggregateException ex)
            {
                exceptions.AddRange(ex.Flatten().InnerExceptions);
            }
            catch (Exception ex) when (!(ex is OutOfMemoryException || ex is StackOverflowException))
            {
                exceptions.Add(ex);
            }
        }

        if (exceptions.Any())
        {
            throw new AggregateException(exceptions);
        }
    }
}

public enum PublishStrategy
{
    /// <summary>
    /// Run each notification handler after one another. Returns when all handlers are finished. In case of any exception(s), they will be captured in an AggregateException.
    /// </summary>
    SyncContinueOnException = 0,

    /// <summary>
    /// Run each notification handler after one another. Returns when all handlers are finished or an exception has been thrown. In case of an exception, any handlers after that will not be run.
    /// </summary>
    SyncStopOnException = 1,

    /// <summary>
    /// Run all notification handlers asynchronously. Returns when all handlers are finished. In case of any exception(s), they will be captured in an AggregateException.
    /// </summary>
    Async = 2,

    /// <summary>
    /// Run each notification handler on it's own thread using Task.Run(). Returns immediately and does not wait for any handlers to finish. Note that you cannot capture any exceptions, even if you await the call to Publish.
    /// </summary>
    ParallelNoWait = 3,

    /// <summary>
    /// Run each notification handler on it's own thread using Task.Run(). Returns when all threads (handlers) are finished. In case of any exception(s), they are captured in an AggregateException by Task.WhenAll.
    /// </summary>
    ParallelWhenAll = 4,

    /// <summary>
    /// Run each notification handler on it's own thread using Task.Run(). Returns when any thread (handler) is finished. Note that you cannot capture any exceptions (See msdn documentation of Task.WhenAny)
    /// </summary>
    ParallelWhenAny = 5,
}

[jizzaki]

I finished implementing this and one recommendation I would make is updating the AddMediatR config to use the Transient lifetime scope.

services.AddMediatR(options => options.Using<CustomMediator>().AsTransient(), assemblies);

The issue I ran into was that when as a Singleton it would not be able to locate the registered assemblies. Not sure why

[fiseni]

Do not use the above implementation, it has flaws. You can find a better implementation here. The NoWait strategies should execute on a separate scope.

Anyhow, starting with 12.0.1 none of this would work. I'll publish an updated version soon.

[jizzaki]

I just ran into that flaw and had to update my side to support having NoWait execute under a new scope via "serviceProvider.CreateScope".

[jizzaki]

I will take a look at the blog. Thanks for your hard work man!

[fiseni]

Yes, you need to create a new scope, but not using injected IServiceProvider (if that's what you're doing). You need a new scope out of the root service provider (e.g. utilizing IServiceScopeFatory).

[alfosua]

Why you don't simply provide a built-in solution?

[jbogard]

Because:

1. The docs are sufficient to implement and extend your own custom solution
2. The docs as-is are nowhere close to what I would want to implement as a built-in solution that supports multiple publish strategies

Namely, the use of an enum is strictly to enable a demo. Folks that want a custom solution will often modify the code to do more complex logic or processing. Unlike behaviors where it was pretty obvious where the extension points could be, custom publishing logic isn't that obvious so I just give you a virtual method to implement whatever you want inside.

[mccow002]

Is there any way to control the order of notifications? Say I want to update my cache first before running all the other events. I would think I could add some flag to the Notification Handler, like IUpdateReadonlyDb, and then look for that in the PublishCore override. But PublishCore only gives you the Handle method as a Func, so I can't inspect the handler class.

[jbogard]

It's completely up to the underlying container on the handlers order. Also - don't. Events/notifications are for discrete handlers that have no process coupling to any others, or the notification raiser. If you need things to happen in a specific order, don't use notifications/events.

[adam-dot-cohen]

Notifications / events are fairly complicated to implement and come with substantial risk of introducing bugs to the battle tested code base - which by the way, hats off - absolutely awesome.

I'm not sure if the way I've wrapped my head around it from the examples is exactly what you intended, but I ended up in the same boat as those on this thread - implementing custom mediators, publishers, etc... Take this with a grain of salt as I have not familiarized myself with the codebase well enough to say the suggestion I'm about to propose is sound. That said, it might be easier for end users to understand and less risky implement if the design pattern was more pain job and less engine tweaks. Again, I'm a laymen on the codebase which is an awesome piece of work so take this Hail Mary for what it is....

What I'll call a "decorator" pattern for lack of a classification seems (again - coming from laymen) to achieve the objective and seems to provide more (transparent) control over logic, sequencing, etc. See the base class below for an untested reference implementation as food for thought - would love to contribute the amazing work you've done or - more likely - be told to sit on the bench with the rest of the amateurs.

/// Base class implementation - PublishAsync is overridden for bespoke publishing implementations.  
/// The method takes bunch of parameters to cover the use cases from your example.  These would 
/// obviously make more sense passed to constructor on config object...
public abstract class PublisherDecoratorBase<TNotification> where TNotification : INotification
{
	protected readonly Context _context;

	public PublisherDecoratorBase(IMediator mediator)
		=> _context = new Context(mediator);
	
	public async Task PublishAsync(TNotification @notification) 
		=> await _context.Notifications.Writer.WriteAsync(@notification);

        protected abstract Task ProcessAsync(bool continueOnException = false, 
		bool throwException = true, CancellationToken token = default);
	
	protected class Context
	{	
		public Context(IMediator mediator) => Mediator = mediator;
		
		// Unbounded to simplify illustration
		public Channel<TNotification> Notifications { get; } = Channel.CreateUnbounded<TNotification>();
		
		public IMediator Mediator {get; init;}
	}
}

Psuedo code implemenation below (never been run) but if you think it's an approach worth pursuing, in the unlikely event you think it's something worth pursuing I'll be more than happy to lend some elbow grease and polish it into actual implementation. The class would obviously be injected into and inherited implementation of Mediator and exposed an overload of Publish

public class AsyncPublisherDecorator<TNotification> : PublisherDecoratorBase<TNotification>
	where TNotification : INotification
{
	public AsyncPublisherDecorator(IMediator mediator) : base(mediator) {}

	// Overriden implementation which handles the use cases in your sample as if I
	// understand them except for parallelizing tasks, which would be handled by simply 
	// running multiple Task.Run(()=>PublishAsync()) calls to add simultaneous consummers
	public override async Task ProcessAsync(bool continueOnException = false,
		bool throwException = true, CancellationToken token = default)
	{
		var exceptions = new List<Exception>();

		await foreach (var item in _context.Notifications.Reader.ReadAllAsync(token))
		{
			try
			{
				await _context.Mediator.Publish(item, token);
			}
			catch (Exception ex)
			{
				if (!continueOnException && throwException) throw;

				if (!continueOnException) break;

				exceptions.Add(ex);
			}
		}
		if (throwException && exceptions.Any())
		{
			throw new AggregateException(exceptions);
		}
	}
}

[aliy007]

Overriding the Mediator and injecting a singleton class with async publish queues made for a straightforward alternative. It also provides a means for high performance FIFO fire-and-forget control flow. Taking it a step further and coupling it with a my existing Redis backplane and a background worker per container has drastically simplified non-pub/sub cloud events as well.