Event bus for CQRS

Question

I write a project with CQRS. I want to build my read model asynchronously from domain events. When i read Greg Young and other topic about CQRS, i see just to use a service bus. But for me a service bus is a technical layer for the message transport and, in this context, the order of processing is not signifiant.

Domain events are messages with a particularity : the order of processing is signifiant.

So, do i use a service bus (NServiceBus, Mass transit...) with an additionnal (complex) layer to ensure the event processing order ? Or a specific "event bus" with this particularity already exists?

Edit

I explain more precisely my situation :

I made a first application with CQRS. In this application, i use a relationnal database for the persistance layer not event sourced. This database is used for the write side and the read side. it's a choice. This application publish domain events on a abstract bus with no implementation.

I need to make a second application with a read model building from the first application.

In this context and if i used an event store, i thing using and event store to build a projection for my second application is a bad thing. If the persistence layer change, the others applications (listeners) are impacted.

Edit 2 : a solution ?

A solution is maybe to use the couple event store / service bus to build an event bus. Thus, the persistance ignorance is preserved and event bus as tools for build projection and listen event.

Edit 3 : GetEventStore

I look at GetEventStore, and it provide a subscribtion mechanism. The solution is maybe use GetEventStore as event bus!

So if i use GetEventStore as event bus, it's means i use event sourcing ? Or event sourcing is a concept only for persistence layer ? Or in my first application i use already event sourcing because i use domain events from aggregate to build the state in a relationnal database ?

So the new question is what is event sourcing ? the storing of events directly in a event store or the using of domain events for build the state ?

Answer 1

So the new question is what is event sourcing ? the storing of events directly in a event store or the using of domain events for build the state ?

"Event sourced" is the property that all of the changes to your domain model are captured in domain events.

Domain events are messages with a particularity : the order of processing is signifiant.

Yes, that's huge to understand; if you are dealing with an event sourced anything, you need to be reading histories (ordered sequences of events), not just the events themselves.

i think to use the event store (in case of choice event sourcing) to make projections for the read model implies a bad dependency with the persistance layer.

Not at all; from the point of view of the read model, there's just a repository that returns histories. The actual implementation details are purely a persistence concern.

So, do i use a service bus (NServiceBus, Mass transit...) with an additionnal (complex) layer to ensure the event processing order

Almost - you use any event publisher you like, with a simple layer to ensure the event processing order.

I want to build my read model asynchronously from domain events.

So your subscriber looks at each event as it goes by, identifying the events that it cares about, and tracks which histories are stale. When it is time to rebuild your model, you reload the histories that have changed (or get the updates to those histories, if you have been keeping track).

You can do a little bit better if your events have version information encoded into them, because you can then track where you are in the stream; and discard any "old" events knowing that you have already processed them.

That is, imagine your subscriber updating an in memory copy of a map, which tracks the latest event in each stream/history/aggregate that it is watching

{ user.7 : 14
, order.14 : 27
, order.12 : 4
}

When it's time to rebuild the read model, you copy that map, and hand it off to the process that will build the model. That process has its own map, describing the versions of the streams that were used to rebuild the previous version.

{ user.7 : 14
, order.14 : 22
, order.12 : 5
}

Here, the process can see that the previous projection was up to date on user.7, is behind on order.14, and ahead on order.12 (maybe the subscriber missed an event, or it hasn't arrived yet, or something).

Since we know the state is state, we rebuild the read model; you could rebuild everything from scratch, or you could notice that only order.14 is out of date, and rebuild from there. Fetch the histories that you need, create the read model, save off the versions of the streams used to create that model, and you are done.

If the first application change his persistence layer, others applications are impacted, isn't it ?

Not necessarily - if you want my data, then you talk to me - not to my persistence layer. You should review what Udi Dahan has to say about services being the technical authority for a [specific business capability][2].

Answer 2

There is a fundamental difference between event-sourcing and messaging.

In a messaging system the messages are largely transient. Some queuing system keep the message around after being pulled from the queue but logically they have been processed and can be forgotten.

For event-sourcing you are explicitly designing the events emanating from the domain model to represent the state of an aggregate. The data in those structures will typically be somewhat different from an inter-system type message that would be sent using a service bus. This is simply due to the fact that you may have more information on the system message than is required to restore the state of an aggregate using event sourcing.

The events in you ES store are always in the correct order. This should be imposed using the version of the aggregate/stream. one user should never "overwrite" another user's events and it shouldn't even be possible.

In order to reconstitute your aggregate you use your ES store. You cannot, and should not, have to rely on service bus-style messages.

To build projections you use your event store. Since these events are in the correct order you do not need anything special to read then. An event store should have some sequence spanning all aggregtes:

| aggregate id | data     | version | sequence number |
| ------------ | -------- | ------- | --------------- |
| 521          | {binary} | 1       | 1               |
| 521          | {binary} | 2       | 2               |
| 521          | {binary} | 3       | 3               |
| 516          | {binary} | 1       | 4               |
| 516          | {binary} | 2       | 5               |
| 7221         | {binary} | 1       | 6               |

Even though the event stream version are per aggregate the sequence number is unique across all. In this way you can read event for a projection starting from sequence number 1.

You can then have as many projections as you need each with some position indicator:

| projection | position |
| ---------- | -------- |
| customer   | 2        |
| order      | 6        |

Your event projection processor will keep reading the events and handing them off to handlers to process. The handlers will update the read model. The position cursor will be moved along.

This is rather handy when you notice that you made some processing error on a projection. You can delete your read model and reset the position back to 0 to have the entire event history re-processed.

Adding a new projection at any time is also very easy.

I hope that helps. I have an event-sourcing mechanism called Shuttle.Recall if you would like to have a look for some inspiration :)

Answer 3

There are a couple of things to consider here.

If you use Event Sourcing, you might want to use GES (geteventstore.com). In this case, events that are used to build the read model don't go via any message bus, but instead retrieved from the event store using subscriptions. In such case you always get a proper order.

If you just have two separate models in "normal" databases, would this be a document or relational database, you probably should use some message bus. There are a few things here too:

• Remember that updating the write side and publishing an event to the message bus should be done in a transaction. Otherwise you risk your read model to be inconsistent

• There is not for NServiceBus or MassTransit as service bus abstractions, to do with message ordering. In fact, you should choose the right transport and it will do the job. For example, AMQP sets some requirements for message ordering and RabbitMQ delivers. Of course, using NSB or MT will make your life much easier in general but it is a completely different thing

• You always can keep your write and read side on the same version by keeping the aggregate version as a field on both sides. In this case you can at least know if you get some event with a gap and decide what to do - wait for the missing event to be delivered or just stop processing and crash to allow manual intervention

Practice shows that all modern brokers support AMQP and follow the standards. You might have duplicates though but this is way less dangerous.

You can choose using ZMQ and this will give you nearly zero latency so the change of delivering messages out of order will be marginal unless your volumes are incredibly high.