Simple SQL Persistence Usage

Component: Sql Persistence | Nuget: NServiceBus.Persistence.Sql (Version: 1.x)
Target NServiceBus Version: 6.x

Prerequisites

MS SQL Server

  1. Ensure an instance of SQL Server Express (Version 2016 or above) is installed and accessible as .\SQLEXPRESS. Create a database SqlPersistenceSample. Or, alternatively, change the connection string to point to different SQL Server instance

MySql

  1. Ensure an instance of MySql (Version 5.7 or above) is installed and accessible as on localhost and port 3306.
  2. Add the username to access the instance to an environment variable named MySqlUserName
  3. Add the password to access the instance to an environment variable named MySqlPassword

Or, alternatively, change the connection string to point to different MySql instance

Code walk-through

This sample shows a Client + Server scenario.

Projects

SharedMessages

The shared message contracts used by all endpoints.

ServerShared

Contains the OrderSaga functionality and is referenced by the Server endpoints

Client

  • Sends the StartOrder message to either EndpointMySql or EndpointSqlServer.
  • Receives and handles the OrderCompleted event.

Servers

  • EndpointMySql and EndpointSqlServer projects act as "servers" to run the saga instance.
  • Receive the StartOrder message and initiate a OrderSaga.
  • OrderSaga requests a timeout with a CompleteOrder data.
  • When the CompleteOrder timeout fires the OrderSaga publishes a OrderCompleted event.

SQL Scripts

Note that only ServerShared has the NServiceBus.Persistence.Sql.MsBuild NuGet package installed. This will cause the following script directories to be populated at build time

  • ServerShared\bin\Debug\NServiceBus.Persistence.Sql\MsSqlServer
  • ServerShared\bin\Debug\NServiceBus.Persistence.Sql\MySql

These scripts will then be copied to the output of both EndpointMySql and EndpointSqlServer and executed at startup.

The endpoints know which scripts to execute via the use of the persistence.SqlVariant(); API usage at configuration time.

Edit
[assembly: SqlPersistenceSettings(
    msSqlServerScripts: true,
    mySqlScripts: true)]

Persistence Config

Configure the endpoint to use SQL Persistence.

MS SQL Server

Edit
var endpointConfiguration = new EndpointConfiguration("Samples.SqlPersistence.EndpointSqlServer");

var persistence = endpointConfiguration.UsePersistence<SqlPersistence>();
var connection = @"Data Source=.\SQLEXPRESS;Initial Catalog=SqlPersistenceSample;Integrated Security=True";
persistence.SqlVariant(SqlVariant.MsSqlServer);
persistence.ConnectionBuilder(
    connectionBuilder: () =>
    {
        return new SqlConnection(connection);
    });

MySql

Edit
var endpointConfiguration = new EndpointConfiguration("Samples.SqlPersistence.EndpointMySql");

var persistence = endpointConfiguration.UsePersistence<SqlPersistence>();
var password = Environment.GetEnvironmentVariable("MySqlPassword");
if (string.IsNullOrWhiteSpace(password))
{
    throw new Exception("Could not extract 'MySqlPassword' from Environment variables.");
}
var username = Environment.GetEnvironmentVariable("MySqlUserName");
if (string.IsNullOrWhiteSpace(username))
{
    throw new Exception("Could not extract 'MySqlUserName' from Environment variables.");
}
var connection = $"server=localhost;user={username};database=sqlpersistencesample;port=3306;password={password};AllowUserVariables=True;AutoEnlist=false";
persistence.SqlVariant(SqlVariant.MySql);
persistence.ConnectionBuilder(
    connectionBuilder: () =>
    {
        return new MySqlConnection(connection);
    });

Order Saga Data

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

Order Saga

Edit
[SqlSaga(
     correlationProperty: nameof(OrderSagaData.OrderId)
 )]
public class OrderSaga :
    SqlSaga<OrderSagaData>,
    IAmStartedByMessages<StartOrder>,
    IHandleTimeouts<CompleteOrder>
{
    static ILog log = LogManager.GetLogger<OrderSaga>();

    protected override void ConfigureMapping(MessagePropertyMapper<OrderSagaData> mapper)
    {
        mapper.MapMessage<StartOrder>(_ => _.OrderId);
    }

    public async 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
        };
        await context.SendLocal(shipOrder)
            .ConfigureAwait(false);

        log.Info("Order will complete in 5 seconds");
        var timeoutData = new CompleteOrder
        {
            OrderDescription = orderDescription
        };

        await RequestTimeout(context, TimeSpan.FromSeconds(5), timeoutData)
            .ConfigureAwait(false);
    }

    public async Task Timeout(CompleteOrder state, IMessageHandlerContext context)
    {
        log.Info($"Saga with OrderId {Data.OrderId} completed");
        var orderCompleted = new OrderCompleted
        {
            OrderId = Data.OrderId
        };
        await context.Publish(orderCompleted)
            .ConfigureAwait(false);
        MarkAsComplete();
    }
}

Related Articles

  • Sagas
    NServiceBus uses event-driven architecture to include fault-tolerance and scalability in long-term business processes.

Last modified