EntityFramework integration

Component: NHibernate Persistence | Nuget: NServiceBus (Version: 6.x)

Prerequisites

  1. Make sure SQL Server Express is installed and accessible as .\SQLEXPRESS.
  2. Create database called nservicebus.
  3. In the database create schemas sender and receiver.

Running the project

  1. Start the Sender project (right-click on the project, select the Debug > Start new instance option).
  2. The text Press <enter> to send a message should be displayed in the Sender's console window.
  3. Start the Receiver project (right-click on the project, select the Debug > Start new instance option).
  4. In the Sender console hit <enter> to send a new message.
In case of exceptions when running the sample, delete tables from the database used by the code (nservicebus). Entity Framework by defult can't update table schemas. If tables use the old schema, the code won't be executed properly.

Verifying that the sample works correctly

  1. The Receiver displays information that an order was submitted.
  2. The Sender displays information that the order was accepted.
  3. Finally, after a couple of seconds, the Receiver displays confirmation that the order has been completed.
  4. Open SQL Server Management Studio and go to the receiver database. Verify that there is a row in saga state table (dbo.OrderLifecycleSagaData), in the orders table (dbo.Orders) and in the shipments table (dbo.Shipments).

Code walk-through

This sample contains three projects:

  • Shared - A class library containing common code including the message definitions.
  • Sender - A console application responsible for sending the initial OrderSubmitted message and processing the follow-up OrderAccepted message.
  • Receiver - A console application responsible for processing the order message.

Sender and Receiver use different schemas in the same database. The database is used for storing technical data by NServiceBus infrastructure, i.e. SQL Server transport and NHibernate persistance, and for storing business data.

Sender project

The Sender does not store any data. It mimics the front-end system where orders are submitted by the users and passed via the bus to the back-end.

Receiver project

The Receiver mimics a back-end system. It is also configured to use SQLServer transport with NHibernate persistence. It uses EntityFramework to store business data (orders and shipments).

Edit
endpointConfiguration.Pipeline.Register(new UnitOfWorkSetupBehaviorBehavior(), "Sets up unit of work for the message");

In order for the Outbox to work, the business data has to reuse the same connection string as NServiceBus persistence:

Edit
public ReceiverDataContext()
    : base("NServiceBus/Persistence")
{
}
public ReceiverDataContext(IDbConnection connection)
    : base((DbConnection) connection, false)
{
}

When the message arrives at the Receiver, TransactionScope is created to ensure consistency of the whole message handling process

  • message is removed from the input queue by the SQL Server transport
  • a new saga instance is created and stored by NHibernate persistence
  • a new Order entity is created
Edit
{
    var order = new Order
    {
        OrderId = message.OrderId,
        Value = message.Value
    };
    context.DataContext().Orders.Add(order);
}
  • a new Shipment entity is created
Edit
{
    var shipment = new Shipment
    {
        OrderId = message.OrderId,
        Location = message.ShipTo
    };
    context.DataContext().Shipments.Add(shipment);
}
  • a reply message is inserted to the queue
  • a timeout request is inserted to the queue

Unit of work

The integration with Entity Framework allows users to take advantage of Unit of Work semantics of Entity Framework's DataContext. A single instance of the context is shared between all the handlers and the SaveChanges method is called after all handlers do their work.

Setting up

The setup behavior makes sure that there is an instance of unit of work wrapper class before the handlers are called.

Edit
public class UnitOfWorkSetupBehaviorBehavior
    : Behavior<IIncomingLogicalMessageContext>
{
    public override async Task Invoke(
        IIncomingLogicalMessageContext context, Func<Task> next)
    {
        var uow = new EntityFrameworkUnitOfWork();
        context.Extensions.Set(uow);
        await next();
        context.Extensions.Remove<EntityFrameworkUnitOfWork>();
    }
}

Creating data context

The data context is created only once, before it is first accessed from a handler.

Edit
class EntityFrameworkUnitOfWork
{
    ReceiverDataContext context;

    public ReceiverDataContext GetDataContext(SynchronizedStorageSession storageSession)
    {
        if (context == null)
        {
            var dbConnection = storageSession.Session().Connection;
            context = new ReceiverDataContext(dbConnection);

            //Don't use transaction because connection is enlisted in the TransactionScope
            context.Database.UseTransaction(null);

            //Call SaveChanges before completing storage session
            storageSession.OnSaveChanges(x => context.SaveChangesAsync());
        }
        return context;
    }
}

The initialization code captures the connection from the NHibernate persistence storage session and registers SaveChangesAsync to be called when the storage session completes successfully (OnSaveChanges).

Related Articles


Last modified