Unit of work when using MassTransit

Question

I'm looking for a way to hook on the message handling pipeline and do some work after a consumer finishes handling some message. My intention is to open a new session and start a transaction(could be done the IoC Container) before handling and disposing them right after it.

In NServiceBus I would use the IMessageModule interface to hook in. does have anything similar to it? Actually disposing the the handler would also do it for, but as I'm using StructureMap as the ObjectBuilder, the Release method just does nothing.

Answer 1

You can register an interceptor to be called before and after each message is consumed. As an example:

        LocalBus = ServiceBusConfigurator.New(x =>
            {
                x.ReceiveFrom("loopback://localhost/mt_client");

                x.BeforeConsumingMessage(() => { _before.Set(); });
                x.AfterConsumingMessage(() => { _after.Set(); });
            });

Take a look at the MessageInterceptor_Specs.cs file in the MassTransit.Tests project for a working unit test.

Answer 2

I went through the same challenge and this is how I went about it. We had an IUnitOfWork and an ITransaction: with Commit() and Rollback() we added a class for TransactionalOperation : ITransaction to added support transactional behaviour to non-db stuff. The IUnitOfWork.Commit() iterates through a list of TransactionalOperations that might have been added to it.

Now to tie the bus to our system: Added an IBus to wrap external bus Implemented a MassTransitBusGateway: IBus then to tie the bus to unit of work: Implemented UnitOfWorkBus: IBus (decorator) - this decorator makes any calls to Publish() to be aware of the unit of work and add it as TransactionalOperation so that Its execution is delayed until the UnitOfWork.Commit()

This way we are abstracting the concrete bus, thus avoiding adding MassTransit dependencies on several projects, (only the clients really needs it) and added transactional behavior to its operations.