NEventStore NServiceBus setup

Question

How do you integrate to NServiceBus when using NEventStore?

I'm new to NSB ans ES and trying to figure out the best setup for NSB when using ES and CQRS.

I'm hooking into NSB the same way as DispatchCommit in example, https://github.com/joliver/EventStore/blob/master/doc/EventStore.Example/MainProgram.cs

1. Do you publish the whole Commit or Commit.Events?
2. Do you create a wrapper around your messages because NSB requires IMessage on your messages? How do you publish to the correct queue then? Because wrapper is generic in contrast to OrderSubmittedEvent for example. If possible I don't want dependency to NSB for my events because then I also have that in my domain.

Some code or guidance is really appreciated.

Answer 1

Here's what I'm using in production:

public sealed class NServiceBusPublisher : IPublishMessages
{
    private const string AggregateIdKey = "AggregateId";
    private const string CommitVersionKey = "CommitVersion";
    private const string EventVersionKey = "EventVersion";
    private const string BusPrefixKey = "Bus.";
    private readonly IBus bus;

    public NServiceBusPublisher(IBus bus)
    {
        this.bus = bus;
    }

    public void Dispose()
    {
        GC.SuppressFinalize(this);
    }

    public void Publish(Commit commit)
    {
        for (var i = 0; i < commit.Events.Count; i++)
        {
            var eventMessage = commit.Events[i];
            var busMessage = eventMessage.Body as IMessage;
            AppendHeaders(busMessage, commit.Headers); // optional
            AppendHeaders(busMessage, eventMessage.Headers); // optional
            AppendVersion(commit, i); // optional
            this.bus.Publish(busMessage);
        }
    }
    private static void AppendHeaders(IMessage message, IEnumerable<KeyValuePair<string, object>> headers)
    {
        headers = headers.Where(x => x.Key.StartsWith(BusPrefixKey));
        foreach (var header in headers)
        {
            var key = header.Key.Substring(BusPrefixKey.Length);
            var value = (header.Value ?? string.Empty).ToString();
            message.SetHeader(key, value);
        }
    }
    private static void AppendVersion(Commit commit, int index)
    {
        var busMessage = commit.Events[index].Body as IMessage;
        busMessage.SetHeader(AggregateIdKey, commit.StreamId.ToString());
        busMessage.SetHeader(CommitVersionKey, commit.StreamRevision.ToString());
        busMessage.SetHeader(EventVersionKey, GetSpecificEventVersion(commit, index).ToString());
    }
    private static int GetSpecificEventVersion(Commit commit, int index)
    {
        // e.g. (StreamRevision: 120) - (5 events) + 1 + (index @ 4: the last index) = event version: 120
        return commit.StreamRevision - commit.Events.Count + 1 + index;
    }
}

Answer 2

In the third version of NSB was introduced feature named Unobtrusive mode. With it you can get rid of NSB dependency in your domain.

This new feature in NServiceBus V3 allows you to pass in your own conventions to determine which types are message definitions instead of using the IMessage, ICommand or IEvent interfaces.