Azure Table Persistence Usage with Transactions

This sample demonstrates a client/server scenario using sagas and outbox persistences to store records atomically by leveraging transactions.

Prerequisites

Ensure that an instance of the latest Azurite Emulator or Azure Cosmos DB Emulator is running.

Projects

SharedMessages

The shared message contracts used by all endpoints.

Client

Sends the StartOrder message to Server.
Receives and handles the OrderCompleted event.

Server

Receives the StartOrder message and initiates an OrderSaga.
OrderSaga sends a ShipOrder message to itself and requests a timeout with an instance of CompleteOrder with the saga data.
ShipOrderHandler handles ShipOrder, stores an OrderShippingInformation record, and publishes an OrderShipped event with a custom header.
OrderSaga receives the OrderShipped event.
OrderSaga publishes an OrderCompleted event when the CompleteOrder timeout is triggered.

Persistence config

Configure the endpoint to use Azure Table Persistence.

var endpointConfiguration = new EndpointConfiguration("Samples.AzureTable.Transactions.Server");
endpointConfiguration.EnableOutbox();

var useStorageTable = true;
var persistence = endpointConfiguration.UsePersistence<AzureTablePersistence>();

var connection = useStorageTable ? "UseDevelopmentStorage=true" :
    "TableEndpoint=https://localhost:8081/;AccountName=AzureTableSamples;AccountKey=C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==";

var tableServiceClient = new TableServiceClient(connection);
persistence.UseTableServiceClient(tableServiceClient);

persistence.DefaultTable("Server");

The saga ID, which is derived from the OrderId, is used as the partition key.

Using Behaviors

The following shows two different ways to provide OrderIds to the saga using behaviors.

Most messages implement IProvideOrderId allowing the OrderId to be used as the partition key.

class OrderIdAsPartitionKeyBehavior : Behavior<IIncomingLogicalMessageContext>
{
    public OrderIdAsPartitionKeyBehavior(IProvidePartitionKeyFromSagaId partitionKeyFromSagaId, ILogger<OrderIdAsPartitionKeyBehavior> logger)
    {
        partitionKeyFromSagaId1 = partitionKeyFromSagaId;
        this.logger = logger;
    }

    public override async Task Invoke(IIncomingLogicalMessageContext context, Func<Task> next)
    {
        var correlationProperty = SagaCorrelationProperty.None;
        if (context.Message.Instance is IProvideOrderId provideOrderId)
        {
            var partitionKeyValue = provideOrderId.OrderId;
            correlationProperty = new SagaCorrelationProperty("OrderId", partitionKeyValue);
        }

        await partitionKeyFromSagaId1.SetPartitionKey<OrderSagaData>(context, correlationProperty);

        if (context.Headers.TryGetValue(Headers.SagaId, out var sagaIdHeader))
        {
            logger.LogInformation("Saga Id Header: {SagaIdHeader}", sagaIdHeader);
        }

        if (context.Extensions.TryGet<TableInformation>(out var tableInformation))
        {
            logger.LogInformation("Table Information: {TableName}", tableInformation.TableName);
        }

        logger.LogInformation("Using partition key '{PartitionKey}'", context.Extensions.Get<TableEntityPartitionKey>().PartitionKey);

        await next();
    }

    public class Registration : RegisterStep
    {
        public Registration() :
            base(nameof(OrderIdAsPartitionKeyBehavior),
                typeof(OrderIdAsPartitionKeyBehavior),
                "Determines the PartitionKey from the logical message",
                provider => new OrderIdAsPartitionKeyBehavior(
                    provider.GetRequiredService<IProvidePartitionKeyFromSagaId>(),
                    provider.GetRequiredService<ILogger<OrderIdAsPartitionKeyBehavior>>()
                    ))
        {
            InsertBefore(nameof(LogicalOutboxBehavior));
        }
    }

    IProvidePartitionKeyFromSagaId partitionKeyFromSagaId1;
    private readonly ILogger<OrderIdAsPartitionKeyBehavior> logger;
}

One handler publishes an event that does not implement IProvideOrderId but adds a custom header containing the OrderId. The handler also creates OrderShippingInformation as part of the transactional batch provided by NServiceBus.

public class ShipOrderHandler (ILogger<ShipOrderHandler> logger):
    IHandleMessages<ShipOrder>
{
    public Task Handle(ShipOrder message, IMessageHandlerContext context)
    {
        var orderShippingInformation = new OrderShippingInformation
        {
            Id = Guid.NewGuid(),
            OrderId = message.OrderId,
            ShippedAt = DateTimeOffset.UtcNow,
        };

        Store(orderShippingInformation, context);

        logger.LogInformation("Order Shipped. OrderId {OrderId}", message.OrderId);

        var options = new PublishOptions();
        options.SetHeader("Sample.AzureTable.Transaction.OrderId", message.OrderId.ToString());

        return context.Publish(new OrderShipped { OrderId = orderShippingInformation.OrderId, ShippingDate = orderShippingInformation.ShippedAt }, options);
    }

    private static void Store(OrderShippingInformation orderShippingInformation, IMessageHandlerContext context)
    {
        var session = context.SynchronizedStorageSession.AzureTablePersistenceSession();

        orderShippingInformation.PartitionKey = session.PartitionKey;

        session.Batch.Add(new TableTransactionAction(TableTransactionActionType.Add, orderShippingInformation));
    }
}

The custom header added then allows the partition key to be determined within OrderIdHeaderAsPartitionKeyBehavior, which derives the same saga ID-based partition key from the OrderId in the header.

class OrderIdHeaderAsPartitionKeyBehavior(IProvidePartitionKeyFromSagaId partitionKeyFromSagaId, ILogger<OrderIdHeaderAsPartitionKeyBehavior> logger) : Behavior<IIncomingLogicalMessageContext>
{
    public override async Task Invoke(IIncomingLogicalMessageContext context, Func<Task> next)
    {
        if (context.Headers.TryGetValue("Sample.AzureTable.Transaction.OrderId", out var orderId))
        {
            var correlationProperty = new SagaCorrelationProperty("OrderId", Guid.Parse(orderId));
            await partitionKeyFromSagaId.SetPartitionKey<OrderSagaData>(context, correlationProperty);

            logger.LogInformation("Found partition key '{PartitionKey}' from header 'Sample.AzureTable.Transaction.OrderId'", context.Extensions.Get<TableEntityPartitionKey>().PartitionKey);
        }

        await next();
    }

    public class Registration : RegisterStep
    {
        public Registration() :
            base(nameof(OrderIdHeaderAsPartitionKeyBehavior),
                typeof(OrderIdHeaderAsPartitionKeyBehavior),
                "Determines the PartitionKey from a header",
                provider => new OrderIdHeaderAsPartitionKeyBehavior(
                    provider.GetRequiredService<IProvidePartitionKeyFromSagaId>(),
                    provider.GetRequiredService<ILogger<OrderIdHeaderAsPartitionKeyBehavior>>()))
        {
            InsertBefore(nameof(OrderIdAsPartitionKeyBehavior));
            InsertBefore(nameof(LogicalOutboxBehavior));
        }
    }
}

Finally, the above behaviors are registered in the pipeline.

endpointConfiguration.Pipeline.Register(new OrderIdHeaderAsPartitionKeyBehavior.Registration());
endpointConfiguration.Pipeline.Register(new OrderIdAsPartitionKeyBehavior.Registration());

Order saga data

public class OrderSagaData :
    ContainSagaData
{
    public Guid OrderId { get; set; }
    public string OrderDescription { get; set; }
}

Order saga

public class OrderSaga(ILogger<OrderSaga> logger) :
    Saga<OrderSagaData>,
    IAmStartedByMessages<StartOrder>,
    IHandleMessages<OrderShipped>,
    IHandleTimeouts<CompleteOrder>
{
    protected override void ConfigureHowToFindSaga(SagaPropertyMapper<OrderSagaData> mapper)
    {
        mapper.MapSaga(saga => saga.OrderId)
            .ToMessage<StartOrder>(msg => msg.OrderId)
            .ToMessageHeader<OrderShipped>("Sample.AzureTable.Transaction.OrderId");
    }

    public Task Handle(StartOrder message, IMessageHandlerContext context)
    {
        var orderDescription = $"The saga for order {message.OrderId}";
        Data.OrderDescription = orderDescription;

        logger.LogInformation("Received StartOrder message {OrderId}. Starting Saga", Data.OrderId);

        var shipOrder = new ShipOrder
        {
            OrderId = message.OrderId
        };

        logger.LogInformation("Order will complete in 5 seconds");
        var timeoutData = new CompleteOrder
        {
            OrderDescription = orderDescription,
            OrderId = Data.OrderId,
        };

        return Task.WhenAll(
            context.SendLocal(shipOrder),
            RequestTimeout(context, TimeSpan.FromSeconds(5), timeoutData)
        );
    }

    public Task Handle(OrderShipped message, IMessageHandlerContext context)
    {
        logger.LogInformation("Order with OrderId {OrderId} shipped on {ShippingDate}", Data.OrderId, message.ShippingDate);
        return Task.CompletedTask;
    }

    public Task Timeout(CompleteOrder state, IMessageHandlerContext context)
    {
        logger.LogInformation("Saga with OrderId {OrderId} completed", Data.OrderId);
        MarkAsComplete();
        var orderCompleted = new OrderCompleted
        {
            OrderId = Data.OrderId
        };
        return context.Publish(orderCompleted);
    }
}