Posts tagged NServiceBus

In the beginning… there were two awesome pieces of code: Ncqrs created by Pieter and Event Store by Jonathan… Now, let me introduce them.

Ncqrs is highly focused on popularizing CQRS and Event Sourcing ideas and, because of that, it is not tied to one particular flavor of CQRS. With Ncqrs you can as well create simple projects backed by RDBMS event store without enforcing aggregate boundaries, as highly scalable reliable solutions without the need of 2PC. The true power of Ncqrs is it’s rich model for representing aggregates. It allows automatic mapping of event handlers using conventions, attributes and via an internal DSL. Ncqrs aggregates can contain entities and nested entities (entities inside entities).

On the other hand, Jonathan’s Event Store is focused on ‘hardcore’ Event Sourcing. The learning curve is high. With Ncqrs you can really write code in the ORMish style. I don’t think you should, but if you can, you have all the facilities, like Unit of Work. You can’t do this with Event Store. This also means that you have smaller chances to create a poorly designed solution. But the feature that makes Event Store a really an awesome tool is support for processing idempotency, integrity and optimistic concurrency based on very loose semantics of most NoSQL databases.

So, why not combine the rich aggregates and asynchronous event processing engine of Ncqrs with storage options (all flavors of SQL, RavenDB, MongoDB), concurrency and idempotency of Event Store? It turned out to be a fairly simple task.

Receiving a command

My first decision was to give up the Ncqrs commanding infrastructure. I wanted to address the scenario where commands are shared resources (e.g. in the enterprise) and they are not allowed to be dependant upon particular framework like Ncqrs (which is an implementation detail of one system). This decision lead me to swap a command service for an interface like this one

public interface IRemoteFacade
{
    void Execute(CommandMetadata metadata, Action<AggregateRoot> commandAction);
}

public class CommandMetadata
{
    public Guid CommandId { get; set; }
    public int? LastKnownRevision { get; set; }
    public Guid? TargetId { get; set; }
    public Type TargetType { get; set; }
}

Instead of forcing command to implement and interface defined by me (or be decorated by my attributes) so that I can handle them in an automatic way, I acknowledge that commands are autonomous beings and that it is handler’s responsibility to extract necessary metadata from the command object. So, why don’t define ICommandHandler interface like this one

public interface ICommandHandler<TCommand>
{
    void Handle(TCommand command);
}

at this level? At first I thought it would be a good idea. Then I realized that this kind of model is already implemented in most transport-level frameworks like NServiceBus. I don’t want to duplicate their functionality. I’d like transport-level handlers (e.g. IHandleMessage<T> in NServiceBus) to call my IRemoteFacade.

Processing the command

After command metadata is extracted and command is handed over to IRemoteFacade, the real processing begins. First, a new instance of aggregate of specified class is created. Then, if the command is targeted against an existing aggregate, an event stream consisting of all events from the latest snapshot up to the last known revision is retrieved. If no revision is provided in the command (which means that optimistic concurrency is disabled for particular command), event are retrieved until most recent commit.

The state of the aggregate is reconstructed using retrieved events. Then, the action provided in Execute call is invoke against the aggregate object.

The last (but certainly not least) activity is saving the state of the aggregate. Events resulting from processing the command are retrieved from the aggregate object, transformed and pushed into the Event Store. If there were optimistic concurrency violations, exception will be thrown. Also, if Event Store detected duplicate command it will signal this by throwing an exception. It is up to IRemoteFacade caller to catch this exception and (most probably) swallow it.

Integration with NServiceBus

All this stuff would be useless if commands cannot be send to the IRemoteFacade. There has to be some transport mechanism. While I don’t want to tie to a single transport, I want to support some common (in my opinion) scenarios. I decided to add NServiceBus support first.

NServiceBus has a notion of message handlers defined as IHandleMessages<T> interface. OOTB you would have to code the facade calling code by hand. To avoid duplication I prepared a special base class:

public abstract class CommandHandlerBase<TMessage, TAggregateRoot> : IHandleMessages<TMessage>
    where TMessage : IMessage
    where TAggregateRoot : AggregateRoot
{
    public IRemoteFacade RemoteFacade { get; set; }
    public IBus Bus { get; set; }

    public virtual void Handle(TMessage message)
    {
        var metadata = ExtractMetadata(message);
        if (metadata.TargetType == null)
        {
            metadata.TargetType = typeof (TAggregateRoot);
        }
        try
        {
            RemoteFacade.Execute(metadata, x => Handle(message, (TAggregateRoot)x));
        }
        catch (DuplicateCommitException ex)
        {
        }
    }

    protected abstract CommandMetadata ExtractMetadata(TMessage message);
    protected abstract void Handle(TMessage message, TAggregateRoot aggregateRoot);
}

The code is straightforward. All you have to do is create your own code for extracting command metadata (or its envelope) and for invoking particular action on the aggregate root. The Handle method is also a good place to put command validation logic. Now, how can I tell NServiceBus to use all this infrastructure? It is as simple as calling one method:

_bus = Configure.With()
    .Log4Net()
    .DefaultBuilder()
    .XmlSerializer()
    .MsmqTransport()
    .IsTransactional(true)
    .PurgeOnStartup(true)
    .UnicastBus()
    .LoadMessageHandlers()
    .InstallNcqrs()
    .UseInMemoryPersistenceEngine()
    .CreateBus()
    .Start();

The additional UseInMemoryPersistenceEngine method instructs Event Store which storage engine to use.

Summary and what’s next

By combining the strongest points of Ncqrs and Event Store I believe I created the most powerful and universal tool for implementing CQRS in .NET platform. Add NServiceBus for command transport and you have a complete end-to-end solution.

The source code can be downloaded from my fork of Ncqrs on github. It is still a prototype but you can see the power it brings.

The next goal is to integrate it smoothly with asynchronous event processing engine of Ncqrs.