NEventStore NServiceBus 设置答案

【问题标题】：NEventStore NServiceBus setupNEventStore NServiceBus 设置
【发布时间】：2011-04-06 09:23:33
【问题描述】：

在使用 NEventStore 时如何与 NServiceBus 集成？

我是 NSB 和 ES 的新手，并试图找出使用 ES 和 CQRS 时 NSB 的最佳设置。

我以与示例中的 DispatchCommit 相同的方式连接到 NSB， https://github.com/joliver/EventStore/blob/master/doc/EventStore.Example/MainProgram.cs

您发布整个 Commit 或 Commit.Events 吗？
您是否因为 NSB 要求您的消息中包含 IMessage 而围绕您的消息创建一个包装器？那么你如何发布到正确的队列呢？因为 wrapper 与 OrderSubmittedEvent 相比是通用的，例如。如果可能的话，我不希望我的事件依赖于 NSB，因为我的域中也有它。

非常感谢一些代码或指导。

【问题讨论】：

标签： nservicebus cqrs neventstore

【解决方案1】：

这是我在生产中使用的：

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;
    }
}

【讨论】：

好的，所以您的域中依赖于 NSB？谢谢乔纳森！
是的，很遗憾。对我来说，“域事件”和您放在网络上的事件之间没有区别。 CQRS 内部对此存在一些争论。

【解决方案2】：

在 NSB 的第三版中引入了名为 Unobtrusive mode 的功能。有了它，您可以摆脱域中的 NSB 依赖。

NServiceBus V3 中的这个新特性允许你传入你自己的确定哪些类型是消息定义的约定使用 IMessage、ICommand 或 IEvent 接口。

【讨论】：