Living in the Tech Avalanche Generation

In a previous post I showed in part what it would take to create a Saga Persister using LINQ To SQL, however it wasn’t the complete story, therefore I decided on creating two new complete persisters to demonstrate how to use the extensibility points provided by NServiceBus to create a persistence mechanism for Sagas. Notice I said two persisters, one for LINQ To SQL and another for the Entity Framework, this post however deals only the LINQ To SQL Persister.

A Place to Start

I think the best place to start is with some typical configuration code that one might use to initialise the ‘BUS’ for a Saga that is using the LINQ To SQL persister. I have included this code twice, once as text (for copy paste) and another as an image with numbered markers to which I will occasionally refer to.

Listing 1.0

static void Main()
{
    LogManager.GetLogger(“hello”).Debug(“Order Started.”);

    try
    {
        const string mappingXml =
            @”..\..\..\OrderService.Persistence\SagaL2SMapping.xml”;
        const string dbConnectionString =
            @”Data Source=BOOMER\BOOM09;” +
            “initial catalog=Sagas;user id=sa;”+
            “password=supremo”;

        var sr = new StreamReader(mappingXml);
        var mapping = sr.ReadToEnd();

        var loadopts = new DataLoadOptions();
        loadopts.LoadWith<OrderSagaData>(o => o.Lines);

        var sessionFactory = LinqToSqlSessionFactory
                    .Configure(mapping, dbConnectionString,
                                loadopts, new DebuggerTextWriter());

        NServiceBus.Configure.With()
            .SpringBuilder(
                (cfg =>
                {
                    cfg.ConfigureComponent<OrderSagaFinder>
                        (ComponentCallModelEnum.Singlecall)
                        .SessionFactory = sessionFactory;
                }))
            .XmlSerializer()
            .MsmqTransport()
                .IsTransactional(true)
                .PurgeOnStartup(true)
            .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);
    }

    Console.Read();
}

Figure 1.0

First up we define a path to the XML mapping file for the entities and a connection string. Yes, I know there are better ways of doing this in production, however I am purposefully making things very plain here for both brevity and clarity. The next thing to note is the use of the LoadOptions class to specify the eager fetching (loading) for any reference data members found in the ISagaEntity implementation for the given Saga. Next up we initialise a LinqToSqlSessionFactory and constructor inject it with it’s dependencies, namely the mapping, connection string, LoadOptions and an optional (can be null), TextWriter for logging SQL output. In order to find Saga’s the NServiceBus infrastructure needs to be configured with a given implementation of the IFindSagas interface, however a ‘Finder’ is not part of the Persister per`se, it does require to be injected with the same mechanism that is responsible for providing a ‘session’ or DataContext in our case. Item 4.0 in Figure 1.0 shows the container “(Spring.Net in this case), configuring our Saga Finder. The last interesting part of the fluent configuration code is the LinqToSqlSagaPersister shown at Item 5.0 in Figure 1.0, which is simply providing the infrastructure the hook to the persister itself. Essentially all that is going on here is that we are providing the NServiceBus infrastructure with what it needs to know to persist a long running services (saga) state, by initialising a Session factory and providing for that Session Factory to be injected into the Saga Persister and Message Module (see page 22) that combine to form all the required moving parts for saga persistence.

The Message Module is an implementation of the NServiceBus IMessageModule will be called at the start and ending of all message handlers and in the case of the Persister, it provides a means to prime and end a Session (DataContext) for the Saga Finder and Persister to manage handle the CRUD aspects of the sagas state.

public class LinqToSqlMessageModule : IMessageModule
{
    public void HandleBeginMessage()
    {
        SessionFactory.GetSession();
    }

    public void HandleEndMessage()
    {
        SessionFactory.CloseSession();
    }

    public virtual LinqToSqlSessionFactory SessionFactory { get; set; }
}

Now for the Persister itself. The Persister implements the ISagaPersister interface(see page 88), and is responsible for (most of) the CRUD operations in managing the persisted state of the Saga. Something to note about the LINQ To SQL Saga Persister that distinguishes itself from the NHibernate Persister that comes out of the box with NServiceBus, is the use of the Generic constructed type. The reason behind the generic is a result of the LINQ To SQL DataContext’s mechanism in retrieving an entity with the GetTable<T>() method.

Listing 2.0

public class LinqToSqlSagaPersister<T> :
    ISagaPersister where T : class, ISagaEntity
{
    private LinqToSqlSessionFactory _sessionFactory;

    public void Complete(ISagaEntity saga)
    {
        var ctx = _sessionFactory.GetSession();
        ctx.GetTable<T>().DeleteOnSubmit(saga as T);
        ctx.SubmitChanges();
    }

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

    public void Save(ISagaEntity saga)
    {
        var ctx = _sessionFactory.GetSession();
        ctx.GetTable<T>().InsertOnSubmit(saga as T);
        ctx.SubmitChanges();
    }

    public void Update(ISagaEntity saga)
    {
        var ctx = _sessionFactory.GetSession();
        ctx.SubmitChanges();
    }

    /// <summary>
    /// The injected Session Factory.
    /// </summary>
    public virtual LinqToSqlSessionFactory SessionFactory
    {
        get { return _sessionFactory; }
        set { _sessionFactory = value; }
    }
}

Here also is a look at a LINQ To SQL finder implemented for the Manufacturing sample that comes with the NServiceBus download. Notice that it’s setup to help find the Saga by whatever relevant fields constitute our search parameters for the messages in our system.

Listing 3.0

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.GetSession()
            .GetTable<OrderSagaData>()
            .Where(o => o.PurchaseOrderNumber == purchaseOrderNumber &&
                o.PartnerId == partnerId)
            .SingleOrDefault();
    }

    private LinqToSqlSessionFactory sessionFactory;

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

Figure 2.0 depicts all the pieces that go to making the LINQ To SQL Persisters library including the ConfigureLinqToSqlSagaPersister.

Figure 2.0

The ConfigureLinqToSqlSagaPersister classes responsibility is to configure the the Persister and Message Module and inject the Session Factory (that provides access to the DataContext) used by both – specifically, opening, closing and persisting the saga’s data to the SQL Server database. One of the extensibility points mentioned and exhibited below in Listing 3.0 is the ConfigureLinqToSqlSagaPersister class, which provides the extension method to the NServiceBus Configure class which is in turn a key part of the configuration fluent interface as demonstrated in Listing 1.0.

Listing 3.0

public static class ConfigureLinqToSqlSagaPersister
{
    /// <summary>
    /// Use the Linq To SQL backed saga persister implementation.
    /// Be aware that this implementation deletes sagas that complete 
    /// so as not to have the database fill up.
    /// Requires that a session factory be configured externally and passed in.
    /// </summary>
    public static Configure LinqToSqlSagaPersister<T>(this Configure config,
        LinqToSqlSessionFactory sessionFactory) where T : class, ISagaEntity
    {
        config.Configurer.ConfigureComponent<LinqToSqlMessageModule>
            (ComponentCallModelEnum.Singleton)
            .SessionFactory = sessionFactory;
        config.Configurer.ConfigureComponent<LinqToSqlSagaPersister<T>>
            (ComponentCallModelEnum.Singlecall)
            .SessionFactory = sessionFactory;

        return config;
    }
}

To round it out we have the Session Factory however as you can see from the class diagram it’s fairly self explanatory and I have made all the code available for download from this link so I suggest you check the Session Factory out for yourself from here. The download code includes the Manufacturing sample already setup to use the LINQ To SQL Saga Persister. Please remember to create a database and name it Sagas, before running the TSQL script in the manufacturing example. Also don’t forget to change the config file settings in the OrderServiceHost project, change the SQL Server Database connection strings where applicable and run the TimeoutManager from the NServiceBus Tools directory. One last thing to note (when you build the solutions): the LinqToSqlSagaPersister.dll is the result of an ILMerge of both the persister and config projects.

I will follow tomorrow with a short post on the Entity Framework Persister, however bear in mind that it will only be applicable to Entity Framework 1.0 so be prepared for some comprises due to the lack of POCO support however I will finish the series with a Persister aimed at the Entity Framework version 4.0 sometime shortly thereafter.