MassTransit and MediatR – Part 1

Introduction

In this post we’ll show how to get started with MassTransit and MediatR in an ASP.NET 6 application. We’ll start out with creating the MassTransit messages and consumers then move on to creating the MediatR requests and handlers and then glue it all together using our IoC container.

You can find the full code here.

What is MassTransit?

MassTransit is a highly popular Open Source messaging library for .NET which greatly simplifies routing messages to handlers (known as consumers) and works with many message brokers such as Azure Service Bus, RabbitMQ. There’s also lots of more advanced features such as sagas, configuration of retries, exponential back-off and a test harness to simplify integration testing etc.

You can find out more about MassTransit here.

With MassTransit, you can also configure your CorrelationId and MessageIds so that you can trace your requests through your distributed system. To see how this can work with ASP.NET APIs, see ASP.NET Correlation ID.

What is MediatR?

MediatR is another Open Source library used to implement the Mediator pattern. The Mediator pattern is a way to decouple code which is great for complex codebases. For example, if you have an API with a controller that needs to call out to a few services, each of these services needs to be injected into the constructor and registered in your IoC container. With MediatR, you can simply inject an IMediator and this will send out requests or notifications which are automatically routed to the right handlers for you. If you need to add additional handlers, you no longer need to add additional services in your IoC container or inject them into your controller.

MediatR also adds pipeline behaviours so you can add validators or loggers into the request pipeline so this helps keep that code out of your handlers where you may have business logic.

You can find out more about MediatR here.

Why use MassTransit with MediatR?

When receiving a message, you may want to perform many actions on this message including validation and logging but you could be writing to a database then calling an API before sending out other messages (after some other transformations). So, if we introduce MediatR, you could receive the message using MassTransit’s consumer which then sends out requests or notifications to multiple MediatR handlers so you can simplify the solution and follow the Single Responsibility Principle to make your code more maintainable.

Getting started with MassTransit and MediatR

In our sample app, we have a Car Booking Service and will be receiving aBookingCreated message from another upstream service. When the message is received, we need to call a Car Booking API through the ICarBookingService.

1) Create initial service from template

Let’s start with creating the initial service using the ASP.NET Web API template in Visual Studio or run the below command if you’re using VSCode or other editor:

dotnet new webapi --name MassTransitMediatR

We can remove WeatherForecast.cs and WeatherForecastController as we won’t need these.

We’ll need to add the MediatR and MassTransit packages next:

• MediatR
• MediatR.Extensions.Microsoft.DependencyInjection
• MassTransit
• MassTransit.Azure.ServiceBus.Core (if you’re using Service Bus like me. For more information on how to integrate with other message brokers, see here)

If you’re going to put your MediatR handlers in a different project (e.g. the application layer) then this also needs to reference the MediatR package but you don’t need MediatR.Extensions.Microsoft.DependencyInjection.

If your MassTransit consumers will be in a different project then you’ll need to add the MassTransit package to that project also.

2) MediatR – Create a command

The way MediatR works is by sending requests and notifications to in memory handlers. It’s a great way to decouple code, especially where you may have many handlers performing actions on a single request or notification. You can learn more about MediatR here.

We’re using CQRS naming – i.e. a command will be used to make a change to a system whereas a query will be used to get data from a system without making further changes. Our MakeCarBookingCommand is below. Note how it implements IRequest – this tells MediatR how to route the command.

using MediatR;

namespace CarBooking.Application.Services.CarBookings.Commands.MakeCarBooking;

public record MakeCarBookingCommand(Domain.Models.CarBooking CarBooking) : IRequest;

2) MediatR – Create a handler

Now that we have our MakeCarBookingCommand above, we need to create a handler to receive this command and action it. As our command implements IRequest, our handler needs to implement IRequestHandler<MakeCarBookingCommand>. This interface requires that we implement the Handle() method. See below (and note that we’re injecting ICarBookingRepository which is what we’re going to use to call our Car Booking API).

using CarBooking.Application.Repositories;
using MediatR;

namespace CarBooking.Application.Services.CarBookings.Commands.MakeCarBooking;

internal class MakeCarBookingCommandHandler : IRequestHandler<MakeCarBookingCommand>
{
    private readonly ICarBookingRepository _carBookingRepository;

    public MakeCarBookingCommandHandler(ICarBookingRepository carBookingRepository)
        => _carBookingRepository = carBookingRepository;

    public async Task<Unit> Handle(MakeCarBookingCommand makeCarBookingCommand, CancellationToken cancellationToken)
    {
        await _carBookingRepository.SendAsync(MapToBooking(makeCarBookingCommand));

        return Unit.Value;
    }

    private static Domain.Models.CarBooking MapToBooking(MakeCarBookingCommand makeCarBookingCommand)
    {
        var carBooking = makeCarBookingCommand.CarBooking;
        return new(carBooking.BookingId, carBooking.FirstName, carBooking.LastName, carBooking.StartDate,
                carBooking.EndDate, carBooking.PickUpLocation, carBooking.Price,
                carBooking.Size, carBooking.Transmission);
    }
}

3) MassTransit – Create a message

MassTransit doesn’t need to have messages implement any particular interface so they are simple plain old C# objects (POCO’s). See the BookingCreated message below. It references other POCOs but we’re not really concerned with those.

Make note that, by default when using Azure Service Bus, the fully qualified name of the message is used to create the topic so it’s best that you don’t change this after deployment.

namespace Contracts.Messages;
public record BookingCreated(string BookingId, BookingSummaryEventData BookingSummary, 
    CarBookingEventData CarBooking, HotelBookingEventData HotelBooking, FlightBookingEventData FlightBooking);

4) Create a MassTransit consumer and inject MediatR

We then create a BookingCreatedConsumer which will be called when a BookingCreated message is received. Note that we’re injecting IMediator into the constructor on line 12. This is going to route our commands to the correct handlers.

Note how the consumer needs to implement IConsumer<BookingCreated>. This tells MassTransit which consumers need to consume which messages. This interface requires that we implement the Consume() method which is called when a BookingCreated message is received.

using CarBooking.Application.Services.CarBookings.Commands.MakeCarBooking;
using Contracts.Messages;
using MassTransit;
using MediatR;

namespace CarBooking.Service.Consumers;

public class BookingCreatedConsumer : IConsumer<BookingCreated>
{
    private readonly IMediator _mediator;

    public BookingCreatedConsumer(IMediator mediator)
        => _mediator = mediator;

    public async Task Consume(ConsumeContext<BookingCreated> context) 
        => await _mediator.Send(new MakeCarBookingCommand(MapToCarBooking(context)));

    private static Domain.Models.CarBooking MapToCarBooking(ConsumeContext<BookingCreated> context)
    {
        var carBooking = context.Message.CarBooking;
        var bookingSummary = context.Message.BookingSummary;

        return new(context.Message.BookingId, bookingSummary.FirstName, bookingSummary.LastName,
            DateTime.UtcNow, DateTime.UtcNow, carBooking.PickUpLocation, 100,
            Enum.Parse<Domain.Enums.Size>(carBooking.Size.ToString()),
            Enum.Parse<Domain.Enums.Transmission>(carBooking.Transmission.ToString()));
    }
}

5) Add MassTransit and MediatR IServiceCollection extensions

Let’s add MassTransit and MediatR to our IoC container. To keep Startup neat and maintainable, let’s add a new class called ServiceCollectionExtensions which will include extension methods for IServiceCollection.

To add MediatR, we’ll create a simply need to create an AddMediatorServices()extension method which calls AddMediatR() and takes a Type which is in the assembly containing our handlers:

public static IServiceCollection AddMediatorServices(this IServiceCollection services)
    {
        services.AddMediatR(typeof(MakeCarBookingCommand));

        return services;
    }

For MassTransit, there’s a bit more work to do depending on what message broker you’re using it with. Here I’m using Azure Service Bus so there’s an AddBus() extension method below where you can see I’m adding MassTransit using AddMassTransit().

We also add the consumers and similar to MediatR, you can simply add the assembly where your consumers are stored (in this case I’m using the assembly where the BookingCreatedConsumer is).

The last thing for MassTransit is to configure the message broker so this is where it’s slightly different for each broker. See how it looks for Azure Service Bus below.

    public static IServiceCollection AddBus(this IServiceCollection services, IConfiguration configuration)
    {
        services.AddMassTransit(o =>
        {
            o.SetKebabCaseEndpointNameFormatter();

            o.AddConsumers(typeof(BookingCreatedConsumer).Assembly);

            o.UsingAzureServiceBus((context, cfg) =>

            {
                cfg.Host(new Uri(configuration["MessageBus:ServiceBusUri"]));
                cfg.ConfigureEndpoints(context);
            });
        });

        return services;
    }

What’s all this talk about kebabs? Well, when MassTransit finds a consumer for a message, it’ll create a entity for it in the message broker e.g. a queue for Azure Service Bus as below. Kebab case is hyphen-separated case. (Note that this is means that all services that listen for BookingCreated will use the same queue so if you’re using pub-sub, that’s not going to work).

The full ServiceCollectionExtensions class looks like this:

using CarBooking.Application.Repositories;
using CarBooking.Application.Services.CarBookings.Commands.MakeCarBooking;
using CarBooking.Infrastructure.Clients;
using CarBooking.Infrastructure.Models;
using CarBooking.Infrastructure.Services;
using CarBooking.Service.Consumers;
using MassTransit;
using MediatR;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;

namespace CarBooking.Service.ServiceCollectionExtensions;

public static class ServiceCollectionExtensions
{
    public static IServiceCollection AddMediatorServices(this IServiceCollection services)
    {
        services.AddMediatR(typeof(MakeCarBookingCommand));

        return services;
    }

    public static IServiceCollection AddServices(this IServiceCollection services, IConfiguration configuration)
    {
        services.AddSingleton<ICarBookingRepository, CarBookingRepository>()
            .AddHttpClient<ICarBookingHttpClient, CarBookingHttpClient>();

        services.AddOptions<CarBookingHttpClientSettings>()
            .Bind(configuration.GetSection(nameof(CarBookingHttpClientSettings)));

        return services;
    }

    public static IServiceCollection AddBus(this IServiceCollection services, IConfiguration configuration)
    {
        services.AddMassTransit(o =>
        {
            o.SetKebabCaseEndpointNameFormatter();

            o.AddConsumers(typeof(BookingCreatedConsumer).Assembly);

            o.UsingAzureServiceBus((context, cfg) =>

            {
                cfg.Host(new Uri(configuration["MessageBus:ServiceBusUri"]));
                cfg.ConfigureEndpoints(context);
            });
        });

        return services;
    }
}

6) Add MassTransit and MediatR to Startup

Let’s now call our extension methods from Startup:

using CarBooking.Service.ServiceCollectionExtensions;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;

namespace CarBooking.Service;

public class Startup
{
    public IConfiguration Configuration { get; }

    public Startup(IConfiguration configuration) => Configuration = configuration;

    public void ConfigureServices(IServiceCollection services)
        => services.AddEndpointsApiExplorer()
            .AddMediatorServices()
            .AddApplicationInsightsTelemetry()
            .AddServices(Configuration)
            .AddBus(Configuration);

    public void Configure()
    {
    }
}

7) Add Service Bus namespace to appsettings.json

Let’s store our Service Bus namespace in appsettings so it can be used by MassTransit. Note that we’re using Azure Managed Identity so we don’t have to inject or store connection strings.

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.AspNetCore": "Warning"
    }
  },
  "MessageBus": {
    "ServiceBusUri": "sb://your-servicebus-namespace.servicebus.windows.net"
  }
}

Conclusion

There you have it. That’s how to add MediatR and MassTransit to an ASP.NET 6 service. We talked about how MediatR and MassTransit works, why you’d want to use them and what benefits you could get by using them together. Obviously, this isn’t going to be required for every service that uses MassTransit but could be very useful to reduce the complexity and keep large projects more maintainable.

You can find the full code here.

See you in the upcoming parts when we’ll talk about how to add retries, validation, dead letter messages and integration tests.

Happy coding!