Living in the Tech Avalanche Generation

public class RequestLeaveSaga : Saga<RequestLeaveSagaData>,
        ISagaStartedBy<IAnnualLeaveRequest>,
        IMessageHandler<IAuthorizationForAnnualLeaveResponse>,
        IMessageHandler<ICancelAnnualLeaveRequest>
{
    public override void Timeout(object state)
    {
        //…timeout logic here
    }

    public void Handle(IAnnualLeaveRequest message)
    {
        Data.Id = Guid.NewGuid();
        Data.LeaveRequestId = message.LeaveRequestId;
        Data.EmployeeId = message.EmployeeId;

        var leaveRequestForAuthorization =
            Bus.CreateInstance<IReqeustingAnnualLeaveAuthorization>(
            m =>
                {
                    m.EmployeeId = message.EmployeeId;
                    m.FirstDayOfLeave = message.LeaveStartsAt;
                    m.NumberOfDaysRequested =
                        Data.CalculateDays(message.LeaveStartsAt,
                            message.LeaveEndsAt);
                    m.TimeOfRequestSubmission = message.TimeOfRequest;
                    m.SagaId = Data.Id;
                });
        Bus.Send(leaveRequestForAuthorization);
    }

    public void Handle(IAuthorizationForAnnualLeaveResponse message)
    {
        //…handles the response message from a HR authorization service
    }

    public void Handle(ICancelAnnualLeaveRequest message)
    {
        //…handles cancel message to invalidate the request for leave
    }
}

try
{
    DataContext sessionFactory = GetContext();

    var bus = NServiceBus.Configure.With()
        .SpringBuilder(
            (cfg =>
            {
                cfg.ConfigureComponent<OrderSagaFinder>
                    (ComponentCallModelEnum.None)
                    .SessionFactory = sessionFactory;
            }))
        .XmlSerializer()
        .MsmqTransport()
            .IsTransactional(true)
            .PurgeOnStartup(false)
        .DbSubscriptionStorage()
            .Table(“Subscriptions”)
            .SubscriberEndpointColumnName(“SubscriberEndpoint”)
            .MessageTypeColumnName(“MessageType”)
        .Sagas()
        .LinqToSqlSagaPersister<OrderSagaData>(sessionFactory)
        .UnicastBus()
            .ImpersonateSender(false)
            .LoadMessageHandlers(
                First<GridInterceptingMessageHandler>
                    .Then<SagaMessageHandler>()
             )
        .CreateBus()
        .Start();
}
catch (Exception e)
{
    LogManager.GetLogger(“hello”).Fatal(“Exiting”, e);
}

.Sagas()
.LinqToSqlSagaPersister<OrderSagaData>(sessionFactory)

Another extensibility point that needs to be addressed in order for the LinqToSqlSagaPersister to show up in the fluent interface API, demands that an implementation of an Extension Method is required for any new custom persister. Let’s look at the Extension method that makes the LinqToSqlSagaPersister available:

namespace NServiceBus
{
    public static class ConfigureLinqToSqlSagaPersister
    {
        public static Configure LinqToSqlSagaPersister<T>(this Configure config,
            DataContext sessionFactory) where T : class, ISagaEntity
        {
            config.Configurer
                .ConfigureComponent<LinqToSqlSagaPersister<T>>
                        (ComponentCallModelEnum.Singlecall)
                .SessionFactory = sessionFactory;

            return config;
        }
    }
}

Out of the box NServiceBus comes with an NHibernate backed Saga Persister that developers can freely use with very predictable and stable results. However, its possible that you may wish to implement your own persister based on your favourite ORM technology or alternate storage.

public class LinqToSqlSagaPersister<T> :
        ISagaPersister where T : class, ISagaEntity
{
    public void Complete(ISagaEntity saga)
    {
        SessionFactory.GetTable<T>().DeleteOnSubmit(saga as T);
        SessionFactory.SubmitChanges();
    }

    public ISagaEntity Get(Guid sagaId)
    {
        return SessionFactory.GetTable<T>()
            .Where(s => s.Id == sagaId).Single();
    }

    public void Save(ISagaEntity saga)
    {
        SessionFactory.GetTable<T>().InsertOnSubmit(saga as T);
        SessionFactory.SubmitChanges();
    }

    public void Update(ISagaEntity saga)
    {
        SessionFactory.SubmitChanges();
    }

    public virtual DataContext SessionFactory { get; set; }
}

<?xml version=”1.0″ encoding=”utf-8″?>
<Database Name=”Sagas”
          xmlns=”http://schemas.microsoft.com/linqtosql/mapping/2007″>
  <Table Name=”dbo.OrderSagaData” Member=”OrderSagaData”>
    <Type Name=”OrderService.OrderSagaData”>
      <Column Name=”Id” Member=”Id” DbType=”UniqueIdentifier NOT NULL”
              IsPrimaryKey=”true” CanBeNull=”false” />
      <Column Name=”Originator” Member=”Originator”
              DbType=”VarChar(50)” CanBeNull=”true” />
      <Column Name=”PurchaseOrderNumber” Member=”PurchaseOrderNumber”
              DbType=”VarChar(50) NOT NULL” CanBeNull=”false” />
      <Column Name=”PartnerId” Member=”PartnerId”
              DbType=”UniqueIdentifier NOT NULL” CanBeNull=”false” />
      <Column Name=”ProvideBy” Member=”ProvideBy”
              DbType=”DateTime NOT NULL” CanBeNull=”false” />
      <Association Name=”OrderSagaData_OrderSagaDataLine” Member=”Lines”
              ThisKey=”Id” OtherKey=”OrderSagaDataId”
              Storage=”_orderLines”/>
    </Type>
  </Table>
  <Table Name=”dbo.OrderSagaDataLines” Member=”OrderSagaDataLines”>
    <Type Name=”OrderService.OrderLine”>
      <Column Name=”Id” Member=”Id” DbType=”UniqueIdentifier NOT NULL”
              IsPrimaryKey=”true” CanBeNull=”false” />
      <Column Name=”OrderSagaDataId” Member=”OrderSagaDataId”
              DbType=”UniqueIdentifier NOT NULL” CanBeNull=”false” />
      <Column Name=”ProductId” Member=”ProductId”
              DbType=”UniqueIdentifier NOT NULL” CanBeNull=”false” />
      <Column Name=”Quantity” Member=”Quantity”
              DbType=”Float NOT NULL” CanBeNull=”false” />
      <Column Name=”AuthorizedQuantity” Member=”AuthorizedQuantity”
              DbType=”Float NOT NULL” CanBeNull=”false” />
      <Association Name=”OrderSagaData_OrderSagaDataLine” Member=”Order”
              ThisKey=”OrderSagaDataId” OtherKey=”Id”
              IsForeignKey=”true” Storage=”_order” />
    </Type>
  </Table>
</Database>

private static DataContext GetContext()
{
    StreamReader sr =
        new StreamReader(“D:\\SagaL2SMapping.xml”);

    //set the mapping source up
    XmlMappingSource mapping = XmlMappingSource.FromStream(sr.BaseStream);

    DataContext db = null;

    //new up a persistance repository
    db = new DataContext(“Data Source=BOOMER\\BOOM09;initial “ +
                        “catalog=Sagas;user id=sa;password=supremo”,
                         mapping);

    var loadopts = new DataLoadOptions();
    loadopts.LoadWith<OrderSagaData>(o => o.Lines);
    db.LoadOptions = loadopts;
    db.DeferredLoadingEnabled = false;

    db.Log = new DebuggerTextWriter();

    return db;
}

public class OrderSagaFinder :
    IFindSagas<OrderSagaData>.Using<OrderMessage>,
    IFindSagas<OrderSagaData>.Using<CancelOrderMessage>
{
    public OrderSagaData FindBy(OrderMessage message)
    {
        return FindBy(message.PurchaseOrderNumber, message.PartnerId);
    }

    public OrderSagaData FindBy(CancelOrderMessage message)
    {
        return FindBy(message.PurchaseOrderNumber, message.PartnerId);
    }

    public OrderSagaData FindBy(string purchaseOrderNumber, Guid partnerId)
    {
        return sessionFactory.GetTable<OrderSagaData>().Where(
            o => o.PurchaseOrderNumber == purchaseOrderNumber &&
                o.PartnerId == partnerId).FirstOrDefault();
    }

    private DataContext sessionFactory;

    public virtual DataContext SessionFactory
    {
        get { return sessionFactory; }
        set { sessionFactory = value; }
    }
}

T FindBy(M message)

I hope this helps if your interested in implementing your own Saga Persister and I recommend that you read Udi’s post on Saga’s and play with the sample from the NServiceBus download to get a better idea about their uses.

For those interested I will be following up soon with an Entity Framework persister in a subsequent post however please note it will only be compatible with V 4.0 of Entity Framework (due to the lack of POCO support in the current version).