This post is part of a series on NServiceBus. In the previous articles, we looked at fundamental concepts for messaging and how to get started with NServiceBus. All the code for this series can be found in the NServiceBusTutorial repository on GitHub.
Unit of Work (UoW) is a pattern that helps manage changes to data across multiple operations, ensuring that all changes are committed or rolled back as a single unit. This is particularly important in distributed systems, where operations may span across repositories, message queues, and notification systems. NServiceBus provides built-in support for the Unit of Work pattern as well, making it easier to implement in your Entity Framework Core applications. In this post, we’ll review the Unit of Work pattern with a simple example focused on database technologies and then explore how NServiceBus enhances this pattern to support transactional messaging.
Introduction 👋
The Unit of Work pattern coordinates multiple operations so they either all succeed or none do. For example, rather than each database repository saving immediately, changes are accumulated and flushed in one atomic commit. If the UoW is committed successfully, all changes are persisted. If the UoW fails, all changes are rolled back. Martin Fowler describes it as a pattern that “keeps track of everything you do during a business transaction that can affect the database. When you’re done, it figures out everything that needs to be done to alter the database as a result of your work.” Often times applications using transactional techniques will use a combination of related patterns including UoW, Repository, and Outbox patterns to achieve or enhance the transactional capabilities of a system. In the .NET ecosystem, many ORMs already implement the UoW pattern, like Entity Framework Core’s DbContext for example.
Simple example implementation 🔨
The following example focuses mainly on database operations and explicitly defines an interface for managing a Unit of Work. Other variants may consider additional concerns such as message publishing or event sourcing in the transaction, but for simplicity, we’ll stick to database operations here.
public interface IUnitOfWork
{
Task BeginTransactionAsync(CancellationToken cancellationToken);
Task CommitTransactionAsync(CancellationToken cancellationToken);
Task RollbackTransactionAsync(CancellationToken cancellationToken);
}
public interface IRepository<T> where T : EntityBase
{
Task<T?> GetByIdAsync(int id, CancellationToken cancellationToken);
Task AddAsync(T entity, CancellationToken cancellationToken);
// ... Other repository methods here
// No CommitAsync/SaveChangesAsync here
}Although we don’t have to use Entity Framework Core to achieve the underlying transaction management, the following is a concrete implementation using it.
public class UnitOfWork : IUnitOfWork
{
private readonly DbContext _dbContext;
private IDbContextTransaction _transaction;
public UnitOfWork(DbContext dbContext)
{
_dbContext = dbContext;
}
public async Task BeginTransactionAsync(CancellationToken cancellationToken)
{
// TODO: possibly check the state of the transaction here
_transaction = await _dbContext.Database.BeginTransactionAsync(cancellationToken);
}
public async Task CommitTransactionAsync(CancellationToken cancellationToken)
{
await _dbContext.SaveChangesAsync(cancellationToken);
await _transaction.CommitAsync(cancellationToken);
}
public async Task RollbackTransactionAsync(CancellationToken cancellationToken)
{
await _transaction.RollbackAsync(cancellationToken);
}
}We can then use these abstractions in an NServiceBus message handler like so, noting that this would look nearly identical if it was called from a non-messaging context, like a domain service or MediatR handler.
public class CreateOrderHandler : IHandleMessages<CreateOrderCommand>
{
private readonly IUnitOfWork _unitOfWork;
private readonly IRepository<Order> _orderRepository;
private readonly IRepository<OrderItem> _orderItemRepository;
public CreateOrderHandler(
IUnitOfWork unitOfWork,
IRepository<Order> orderRepository,
IRepository<OrderItem> orderItemRepository)
{
_unitOfWork = unitOfWork;
_orderRepository = orderRepository;
_orderItemRepository = orderItemRepository;
}
public async Task Handle(CreateOrderCommand message, IMessageHandlerContext context)
{
var order = new Order
{
// Map properties from the command to the Order entity
};
await _unitOfWork.BeginTransactionAsync(cancellationToken);
try
{
await _orderRepository.AddAsync(order, cancellationToken);
foreach (var item in order.Items)
{
await _orderItemRepository.AddAsync(item, cancellationToken);
}
await _unitOfWork.CommitTransactionAsync(cancellationToken);
}
catch
{
await _unitOfWork.RollbackTransactionAsync(cancellationToken);
throw;
}
}
}In this example, the CreateOrderHandler uses the IUnitOfWork to manage a transaction that encompasses multiple repository operations. The transaction is started with BeginTransactionAsync, and if all operations succeed, it is committed with CommitTransactionAsync. If any operation fails, the transaction is rolled back with RollbackTransactionAsync, ensuring that no partial changes are persisted.
This code works, but since it relies on manually managing the transaction it can become cumbersome as the complexity of the operations increases and bloats the intended purpose of this code with infrastructure concerns. This becomes even more true as we send NServiceBus messages within the same transaction, especially when our NServiceBus applications should be using Outbox pattern (we’ll look at why in an upcoming blog post this fall 🎃). Luckily, NServiceBus provides a way to simplify this entire layer of what I would call “plumbing code” and easily tie Entity Framework Core’s transactions to NServiceBus’s messaging transaction.
Taking things to the next level 🚀
NServiceBus can be configured to manage the transaction scope automatically, allowing you to focus on your business logic without worrying about the underlying transaction management. Since we’re using Entity Framework Core, we simply need to tie the DbContext’s transaction to NServiceBus’s message processing pipeline. Luckily, we can do this by telling NServiceBus to call EF Core’s SaveChangesAsync at the same time it commits the messages.
While we’re exploring NServiceBus’s capabilities, let’s update the above sample code to send a OrderCreatedEvent at the end of the handler to demonstrate this integration.
// ...
try
{
await _orderRepository.AddAsync(order, cancellationToken);
foreach (var item in order.Items)
{
await _orderItemRepository.AddAsync(item, cancellationToken);
}
await context.Publish(new OrderCreatedEvent(order.Id));
await _unitOfWork.CommitTransactionAsync(cancellationToken);
}
// ... rest of the handlerAssuming persistence has already been configured for your NServiceBus endpoint, the hook for connecting NServiceBus to Entity Framework Core is relatively straightforward:
var endpointConfiguration = new EndpointConfiguration("YourEndpointName");
endpointConfiguration.RegisterComponents(c =>
{
c.AddScoped(b =>
{
var session = b.GetRequiredService<ISqlStorageSession>();
// Create the DbContext with the session's connection
var context = new AppDbContext(session.Connection);
// Use the same underlying transaction, tying NServiceBus to EF Core
context.Database.UseTransaction(session.Transaction);
// Ensure changes are saved when the session is committed
session.OnSaveChanges((s, cancellationToken) => context.SaveChangesAsync(cancellationToken));
return context;
});
});This call to RegisterComponents effectively ties the lifecycle of the Entity Framework Core DbContext to the NServiceBus message processing pipeline, ensuring that the same database transaction is used for both the message handling and the EF Core operations. This is where the 🪄 magic 🪄 really happens and is one of my favorite features of using NServiceBus with EF Core. You can view this NServiceBus sample for enabling Unit of Work with other database technologies.
The CreateOrderHandler can then be refactored to remove the IUnitOfWork dependency since we are delegating that responsibility to the NServiceBus pipeline.
public class CreateOrderHandler : IHandleMessages<CreateOrderCommand>
{
private readonly IRepository<Order> _orderRepository;
private readonly IRepository<OrderItem> _orderItemRepository;
public CreateOrderHandler(
IRepository<Order> orderRepository,
IRepository<OrderItem> orderItemRepository)
{
_orderRepository = orderRepository;
_orderItemRepository = orderItemRepository;
}
public async Task Handle(CreateOrderCommand message, IMessageHandlerContext context)
{
var order = new Order
{
// Map properties from the command to the Order entity
};
await _orderRepository.AddAsync(order, cancellationToken);
foreach (var item in order.Items)
{
await _orderItemRepository.AddAsync(item, cancellationToken);
}
await context.Publish(new OrderCreatedEvent(order.Id));
}
}This handler is now significantly easier to understand and mostly focuses on the business process of creating an order, without being concerned about transaction management. We are able to focus on data operations without having to make explicit calls to SaveChangesAsync per repository and send messages without interacting with our messaging infrastructure. Since our Unit of Work is configured at the endpoint-level, this allows all our message handlers in the project to benefit from the UoW pattern. It’s been my experience that enabling this pattern leads to more maintainable and testable code as well as a much more pleasant development experience.
How Does This Work? 🧐
The previous code sample leverages the configuration we defined to tie NServiceBus to Entity Framework Core’s transaction management. When NServiceBus processes a message, it automatically creates a transaction scope that includes the database operations performed within the handler. So behind the scenes, the pseudo-code for handling this message might look something like this:
// .. Do messaging stuff like read off the queue, deduplication, etc
// Get the persistence configuration from a DI container
var persistence = GetPersistence();
// This may vary depending on your database provider
var transaction = new Transaction(persistence.Connection);
transaction.Begin();
try
{
// Call the message handler
await handler.Handle(message, context);
// Write messages to the Outbox AND call SaveChangesAsync()
// on EF Core's DbContext in the same atomic commit
transaction.Commit();
}
catch
{
// If either messages or domain data fails to write, rollback everything
transaction.Rollback();
throw;
}
// Finishing message processingA Note about send-only endpoints ➡️
If you are using NServiceBus in a send-only endpoint (i.e., an endpoint that only sends messages and does not handle incoming messages), you can still leverage the Unit of Work pattern by using the ITransactionalSession interface. We’ll explore this in more detail in a future post.
Conclusion 🧠
In this blog post, we explored how to simplify the Unit of Work pattern in an NServiceBus application using Entity Framework Core. By leveraging the built-in transaction management capabilities of NServiceBus, we were able to eliminate the need for an explicit Unit of Work implementation, resulting in cleaner and more maintainable code.
The key takeaways from this approach are:
-
Transaction Management: NServiceBus automatically manages database transactions for you, ensuring that all operations within a message handler are executed within a single transaction scope.
-
Simplified Handlers: By configuring NServiceBus and Entity Framework Core together, we can focus on the business logic without worrying about transaction boundaries.
-
Consistency: This approach helps maintain consistency between the messaging and data access layers, reducing the risk of data inconsistencies.
Overall, this pattern can greatly enhance the development experience when working with NServiceBus and Entity Framework Core, allowing developers to build robust and reliable applications with ease.