MediatR’s Pipelines

The real power of MediatR

The library introduces an easy way to augment the behavior of our request handlers via Pipelines:

• IPipelineBehaviors (something that is executed before and after our handler)
• IRequestPreProcessors (something that is executed before our handler)
• IRequestPostProcessors (something that is executed after our handler)

Something that we all do is build incremental code, in a simple CRUD, so first we deliver the database operations (Create/Update/Delete/Read).

Eventually we realize a little more is required, ie: Once we create the User we want to inform another app in the company that the user was created via Service Bus.

Why MediatR makes our life easier? Because we don't need to change the code!

What??? How???

Our deployed code in production doesn’t have to change, our handlers are working and we want to keep them that way.

We don’t want to change all the tests, we don’t want to add any more mocks, because that’s work that can lead to errors.

So Pipelines to the rescue!

Show me the code!

IRequestPostProcessors

• First we declare the action we want to execute as a IRequestPostProcessor

public class ServiceBusPostProcessor : IRequestPostProcessor<CreateUser, User>
{
    private readonly IServiceBus _serviceBus;
    
    public ServiceBusPostProcessor(IServiceBus serviceBus) => _serviceBus = serviceBus;
    
    public Task Process(CreateUser request, User response, CancellationToken cancellationToken)
    {
        _serviceBus.Send(new UserCreated() { User = user });
    }
}

• Finally we need to register the PostProcessor (the automagic function we have learn to use to register MediatR and handlers WILL NOT register Pipelines)

services.AddScoped<IRequestPostProcessor<CreateUser, User>, ServiceBusPostProcessor>;

That’s it, MediatR will take care of everything, so after the IMediator calls the Handle method of the RequestHandler will execute each IRequestPostProcessor<CreateUser, User> registered.

• The order of registration matters!

services.AddScoped<IRequestPostProcessor<CreateUser, User>, ServiceBusPostProcessor>;
services.AddScoped<IRequestPostProcessor<CreateUser, User>, AuditTrailPostProcessor>;

In this case the ServiceBusPostProcessor will be executed before the AuditTrailPostProcessor.

IRequestPreProcessors

Now… What if we want to add some validation to our command? Yes, you guessed correctly, we define an IRequestPreProcessor<CreateUser,User>

public class ValidationPreProcessor : IRequestPreProcessor<CreateUser>
{               
    public Task Process(CreateUser request, CancellationToken cancellationToken)
    {
        if(string.IsNullOrEmpty(request.User.Name))
        {
            throw new ArgumentNullException("name");
        }
    }
}

• Let’s register the PreProcessor (remember: the automagic function we have learn to use to register MediatR and handlers WILL NOT register Pipelines)

services.AddScoped<IRequestPreProcessor<CreateUser>, ValidationPreProcessor>;

Notice the difference between the PreProcessor and PostProcessor?

The PostProcessor receives in the Process method the result of the Handle of the Request Handler, because ideally we want to do something with it.

In the PreProcessor on the other hand, when the Process is called we haven’t reached yet to the Handle method of the Request Handler.

IPipelineBheaviors

Caching made easy!

We have seen how is easy is to enrich out command dispatchers without changing 1 single line of code of the handlers….

Now we will implement caching for a Request command

If we have the Request

public class GetUserById : IRequest<User>
{
    public GetUserById( Guid id ) => Id = id;
    public Guid Id { get; }
}

And our handler

public class QueriesHandler : IRequestHandler<GetUserById, User>
{
    private readonly Database _database;
    
    public QueriesHandler(Database database) _database = database;
    
    public Task<User> Handle(GetUserById request, CancellationToken cancellationToken)
    {
        return _database.Users.FirstOrDefaultAsync(x => x.Id == request.Id, cancellationToken);
    }
}

A simple Cache implemented with MediatR

public class Cache : IPipelineBheaviors<GetUserById, User>
{
    private readonly ICache _cache;
    
    public Cache(ICache cache) _cache = cache;
    
    public async Task<User> Handle(GetUserById request, CancellationToken cancellationToken, RequestHandlerDelegate<User> next)
    {
        if(_cache.TryGetValue(request.Id, out var user))
        {
            return user;
        }
        
        user = await next();
        
        if(user != null)
        {
            _cache.Save(request.Id, user);
        }
        
        return user;
    }
}

The most important thing here is the RequestHandlerDelegate<User> that is nothing more than the next component of our pipeline to be executed and it could be

• an IRequestPreProcessor
• The Request Handler

We are not much interested to know who it is…. All we know is that if we call it, we will receive the result.

This is real power of MediatR, how it helps to deliver new functionality without modifying any deployed code.

We have seen in the last years the increased use of Middleware at ASP.Net Core and the wide adoption of it, nonetheless I have seen many developers reluctant to use the same mechanismn for their command execution.

Happy coding!