MassTransit and MediatR – Part 3

Introduction

In this post, we’ll look at how we can get MassTransit to dead letter invalid messages. We may receive a message and it’s invalid but we don’t want to just throw an exception and retry the message till all the retries are exhausted as the message will never be processed. With Service Bus, MassTransit won’t technically dead letter an invalid message but we can get it to move it to an error queue which has the same effect. We’ll use MediatR pipeline behaviours, FluentValidation and MassTransit so it’ll be an interesting post.

In part 1 of MassTransit and MediatR, we looked at the initial set up and in part 2, we looked at how we can configure MassTransit to do exponential retries when working with Service Bus. If you missed it, you can go back to that post here.

You can find the full working code here.

MassTransit and MediatR and FluentValidation

Three different .NET libraries. Somewhat unrelated. Let’s talk through these and how they fit in this solution.

MassTransit error handling

As we saw in the last post, we can configure the retries when there is an exception thrown when handling a message. One thing that MassTransit enables is ignoring particular exceptions – i.e. we can configure it to not retry in case of a particular exception and this is exactly what we’re going to do – we’re going to throw a ValidationException and configure MassTransit to not retry. You can find out more about this feature here.

FluentValidation

FluentValidation is a great library that helps you write validations which are easy to read and modify. It’s far better than lots of if..else statements. FluentValidation also enables you to have validators in different files and you can register them in your IoC container and inject them where needed.

MediatR Pipeline Behaviours

Rather than putting guard clauses or some other form of validation logic in the MediatR handlers, we can use MediatR Pipeline Behaviours. These Pipeline Behaviours get executed before a MediatR Request or Notification is processed by the handler so that means you can separate out that code from the actual handler so you have cleaner code and less going on in the handlers. If you want, you could have multiple Pipeline Behaviours so you could have one for logging also. It’s essentially the decorator pattern for handlers.

How to configure MassTransit and MediatR to dead letter invalid messages

Add FluentValidation validator

First, we need to add the FluentValidation package. Carrying on from where we left off in Part 1, we are receiving the BookingCreated message and then using MediatR to send a MakeCarBookingCommand which is received by our MakeCarBookingCommandHandler. We can then make a validator for the MakeCarBookingCommand as below. It’s simply checking that the BookingId is 4 characters long:

using FluentValidation;

namespace CarBooking.Application.Services.CarBookings.Commands.MakeCarBooking;
public class MakeCarBookingCommandValidator : AbstractValidator<MakeCarBookingCommand>
{
    public MakeCarBookingCommandValidator() 
        => RuleFor(v => v.CarBooking.BookingId).MinimumLength(4).WithMessage("BookingId cannot be empty");
}

To add the validators to our IoC container, we can use the IServiceCollection extension method that FluentValidation provides for us as below. This essentially registers all validator classes it finds. You can find out more about this here.

services.AddValidatorsFromAssembly(typeof(MakeCarBookingCommandValidator).Assembly);

You can use any Type that’s in the assembly where your validators are.

Add MediatR Pipeline Behaviour

MediatR Pipeline Behaviours need to implement IPipelineBehavior<TRequest, TResponse> according to the documentation here. So, what we’ll do is inject the FluentValidation validators we have created for that particular IRequest (see above) and we’ll basically call each validator. If there are any validation failures, we store the errors in a list which we then pass to the ValidationException when we throw. If there’s no validation errors, we simply call the next item in the pipeline. You’ll find this is similar to the way middleware works in ASP.NET.

Here’s our pipeline behaviour called RequestValidationBehaviour:

using FluentValidation;
using MediatR;

namespace CarBooking.Application.Common.Behaviours;

public class RequestValidationBehaviour<TRequest, TResponse> : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    private readonly IEnumerable<IValidator<TRequest>> _validators;

    public RequestValidationBehaviour(IEnumerable<IValidator<TRequest>> validators) => _validators = validators;

    public Task<TResponse> Handle(TRequest request, CancellationToken cancellationToken,
        RequestHandlerDelegate<TResponse> next)
    {
        var context = new ValidationContext<TRequest>(request);

        var failures = _validators
            .Select(v => v.Validate(context))
            .SelectMany(r => r.Errors)
            .Where(r => r != null)
            .ToList();

        return failures.Count > 0
            ? throw new ValidationException(failures)
            : next();
    }
}

Let’s go ahead and configure our IoC container to inject the pipeline behaviours:

services.AddTransient(typeof(IPipelineBehavior<,>), typeof(RequestValidationBehaviour<,>));

Configure MassTransit to dead letter invalid messages

Now, the final part is to configure MassTransit to not retry in case of an invalid message. If you remember back to part 2 where we configured exponential retries, here we’re going to simply modify that retry policy so that it doesn’t retry if a ValidationException is thrown:

        retryConfigurator.Exponential(int.MaxValue, TimeSpan.FromMilliseconds(200), TimeSpan.FromMinutes(120), 
                TimeSpan.FromMilliseconds(200))
            .Ignore<ValidationException>();

Test MassTransit dead lettering messages

Okay, so let’s try this out. I’m simply going to send through a message which doesn’t have a BookingId and see what happens:

                                {
  "messageId": "7217a64b-ea82-4c28-ad4b-0057ff787d04",
  "requestId": null,
  "correlationId": "7217a64b-ea82-4c28-ad4b-0057ff787d04",
  "conversationId": "6ca30000-4d41-18c0-ffc5-08da5437ea2a",
  "initiatorId": null,
  "sourceAddress": "sb://myservicebus.servicebus.windows.net/MyComputer_WebBffApi_bus_p1toyynperccyegsbdpfep7bdo?autodelete=300",
  "destinationAddress": "sb://myservicebus.servicebus.windows.net/Contracts.Messages/BookingCreated",
  "responseAddress": null,
  "faultAddress": null,
  "messageType": [
    "urn:message:Contracts.Messages:BookingCreated",
    "urn:message:Contracts.Messages:BookingEventBase"
  ],
  "message": {
    "bookingId": "",
    "bookingSummary": {
      "firstName": "Joe",
      "lastName": "Bloggs",
      "startDate": "2024-01-12T15:28:33.6327081Z",
      "endDate": "2024-01-12T15:28:33.6327081Z",
      "destination": "France",
      "price": 574
    },
    "carBooking": {
      "pickUpLocation": "Paris",
      "size": 2,
      "transmission": 0
    },
    "hotelBooking": {
      "numberOfBeds": 2,
      "breakfastIncluded": true,
      "lunchIncluded": true,
      "dinnerIncluded": true
    },
    "flightBooking": {
      "outboundFlightTime": "2024-01-12T15:28:33.6327081Z",
      "outboundFlightNumber": "2b278",
      "inboundFlightTime": "2024-01-12T15:28:33.6327081Z",
      "inboundFlightNumber": "2b279"
    }
  },
  "expirationTime": null,
  "sentTime": "2022-06-22T10:13:55.4966571Z",
  "headers": {},
  "host": {
  "machineName": "MyComputer",
    "processName": "WebBff.Api",
    "processId": 41836,
    "assembly": "WebBff.Api",
    "assemblyVersion": "1.0.0.0",
    "frameworkVersion": "6.0.6",
    "massTransitVersion": "8.0.2.0",
    "operatingSystemVersion": "Microsoft Windows NT 10.0.22000.0"
  }
}

When the message is received, we get a ValidationException thrown by the RequestValidationBehaviour to say that the BookingId can’t be empty:

After this, we can see that MassTransit creates a new booking-created_error queue if it didn’t already exist and then we find our message enters that queue. Although it’s not technically the formal Service Bus dead letter queue, the message is now in error state and won’t be retried so it has the same effect:

Conclusion

In this post, we’ve done a walk through of how to configure MassTransit to dead letter an invalid message using FluentValidation and MediatR Pipeline Behaviours. Stay tuned for the next post where we’ll do a bit of a walkthrough how to test our service using some clever features of MassTransit.

You can find the full working code here.

Happy Coding!