This sample demonstrates a client/server scenario using saga and outbox persistences to store records atomically by leveraging transactions. The Saga ID is used as a partition key.
Projects
SharedMessages
The shared message contracts used by all endpoints.
Client
- Sends the
StartOrdermessage toServer. - Receives and handles the
OrderCompletedevent.
Server projects
- Receive the
StartOrdermessage and initiate anOrderSaga. OrderSagarequests a timeout with an instance ofCompleteOrderwith the saga data.- Receive the
OrderShippedmessage with a custom header. OrderSagapublishes anOrderCompletedevent when theCompleteOrdertimeout fires.
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");
Using Behaviors
Most messages implement IProvideOrderId. Since Saga IDs are deterministically derived from saga data (the correlation property name and value), they can be used as a partition key. Then, IProvidePartitionKeyFromSagaId can be injected into behaviors in the logical pipeline stage.
class OrderIdAsPartitionKeyBehavior : Behavior<IIncomingLogicalMessageContext>
{
public OrderIdAsPartitionKeyBehavior(IProvidePartitionKeyFromSagaId partitionKeyFromSagaId)
{
partitionKeyFromSagaId1 = partitionKeyFromSagaId;
}
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))
{
Log.Info($"Saga Id Header: {sagaIdHeader}");
}
if (context.Extensions.TryGet<TableInformation>(out var tableInformation))
{
Log.Info($"Table Information: {tableInformation.TableName}");
}
Log.Info($"Found partition key '{context.Extensions.Get<TableEntityPartitionKey>().PartitionKey}' from '{nameof(IProvideOrderId)}'");
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>()))
{
InsertBefore(nameof(LogicalOutboxBehavior));
}
}
IProvidePartitionKeyFromSagaId partitionKeyFromSagaId1;
static readonly ILog Log = LogManager.GetLogger<OrderIdAsPartitionKeyBehavior>();
}
Even though one of the handlers replies with a message that does not implement IProvideOrderId, transactionality is still maintained because messages that are part of a saga conversation flow automatically have the Saga ID set as a header. In these cases, there is no need to extract correlation property information to derive the Saga ID.
Finally the above behavior is registered in the pipeline.
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 :
Saga<OrderSagaData>,
IAmStartedByMessages<StartOrder>,
IHandleMessages<OrderShipped>,
IHandleTimeouts<CompleteOrder>
{
static readonly ILog Log = LogManager.GetLogger<OrderSaga>();
protected override void ConfigureHowToFindSaga(SagaPropertyMapper<OrderSagaData> mapper)
{
mapper.MapSaga(saga => saga.OrderId)
.ToMessage<StartOrder>(msg => msg.OrderId);
}
public Task Handle(StartOrder message, IMessageHandlerContext context)
{
var orderDescription = $"The saga for order {message.OrderId}";
Data.OrderDescription = orderDescription;
Log.Info($"Received StartOrder message {Data.OrderId}. Starting Saga");
var shipOrder = new ShipOrder
{
OrderId = message.OrderId
};
Log.Info("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)
{
Log.Info($"Order with OrderId {Data.OrderId} shipped on {message.ShippingDate}");
return Task.CompletedTask;
}
public Task Timeout(CompleteOrder state, IMessageHandlerContext context)
{
Log.Info($"Saga with OrderId {Data.OrderId} completed");
MarkAsComplete();
var orderCompleted = new OrderCompleted
{
OrderId = Data.OrderId
};
return context.Publish(orderCompleted);
}
}