Transaction Configuration

Transactions, and using a shared transaction, is an advanced concept. Every scenario is different, so this is more of a guideline than a rule.

The message pipeline in MassTransit is asynchronous, leveraging the Task Parallel Library (TPL) extensively to maximize thread utilization. This means that receiving an individual message may involve several threads over the life cycle of the consumer. To prevent strange things from happening, developers should avoid using any static or thread static variables as these are one of the main causes of errors in asynchronous programming.

Caution

The message pipeline is asynchronous. Using TransactionScope in an asynchronous context can lead to unexpected behavior. The transaction middleware is preferred over TransactionScope for message consumers.

Configure a transaction filter

MassTransit includes transaction middleware to share a single committable transaction across any number consumers and any dependencies used by those consumers. To use the middleware, it must be added to the receive endpoint.

services.AddMassTransit(x =>
{
    x.AddConsumer<UpdateCustomerAddressConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.ReceiveEndpoint("event_queue", e =>
        {
            e.UseTransaction(x =>
            {
                x.Timeout = TimeSpan.FromSeconds(90);
                x.IsolationLevel = IsolationLevel.ReadCommitted;
            });

            e.ConfigureConsumer<UpdateCustomerAddressConsumer>(context);
        });
    });
});

For each message, a new CommittableTransaction is created. This transaction can be passed to classes that support transactional operations, such as DbContext, SqlCommand, and SqlConnection. It can also be used to create any TransactionScope that may be required to support a synchronous operation.

To use the transaction directly in a consumer, the transaction can be pulled from the ConsumeContext.

public class TransactionalConsumer :
    IConsumer<UpdateCustomerAddress>
{
    readonly SqlConnection _connection;

    public async Task Consume(ConsumeContext<UpdateCustomerAddress> context)
    {
        var transactionContext = context.GetPayload<TransactionContext>();

        _connection.EnlistTransaction(transactionContext.Transaction);

        using (SqlCommand command = new SqlCommand(sql, _connection))
        {
            using (var reader = await command.ExecuteReaderAsync())
            {
            }
        }
    }
}

The connection (and by use of the connection, the command) are enlisted in the transaction. Once the method completes, and control is returned to the transaction middleware, if no exceptions are thrown the transaction is committed (which should complete the database operation). If an exception is thrown, the transaction is rolled back.

Transaction options

Option	Description
Timeout	The maximum time to wait for the transaction to complete
IsolationLevel	The transaction isolation level (default: ReadCommitted)

Transaction guidance

• A class that provides the connection, and enlists the connection upon creation, should be added to the container to ensure that the transaction is not enlisted twice.
• As long as only a single connection string is enlisted, the DTC should not get involved.
• Using the same transaction across multiple connection strings is a bad thing, as it will make the DTC come into play which slows the world down significantly.