CQRS + Event sourcing - event groups and event propagation

Question

This is my current CQRS use-case setup:

Command DTO is received in application layer handler where we map Command DTO to appropriate domain objects if needed, re-hydrate aggregate root from repo and call some method on AR

class CommandHandler{
    handle(Command command){
        Dom1 dom1 = new Dom1(command.d1);
        Dom2 dom2 = new Dom2(command.d2);
        AggregateRoot ar = repo.rehydrate(command.arId);
        ar.doSmth(dom1, dom2, command.int1);
    }
}

Inside doSmth method in AR, I don't have immediately event to apply, but I have to pass some args to entity that is part of this AR. So inside doSmth method I have
```
class AggregateRoot{
    void doSmth(Dom1 dom1, Dom2 dom2, Integer int1){
        subEnt = new OrherSubEntity();
        subEnt.doSmth(int1);
    }
}
```

Now I have sub entity that is activated and generates one DomainEvent. This domain event should be part of event stream that is saved in EventStore and at same time, it is propagated to any listener in this aggregate root.

class SubEntity{
    void doSmth(Integer int1){
        //validate int1
        apply(new SubEntityEvent(int1));
    }

     void when(SubEntityEvent event){
        //just modify local fields
        //used also in event sourcing rehydration
     }
}

Since second sub entity that is listening to SubEntityEvent is now activated, it performs CPU intense operation and generates SomeCPUIntenseCalcHappened DomainEvent. This event should also be part of the event stream that is saved in EventStore. It is also propagated to other sub-entities in same AR but since nobody else listens to it then we have no further processing.
```
class OtherSubEntity{
    void listen(SubEntityEvent event){
        data = SomeCPUIntenseCalc();
        apply(SomeCPUIntenseCalcHappened(data));
    }

    void when(SomeCPUIntenseCalcHappened event){
        //just modify local fields
        //used also in event sourcing rehydration
    }
}
```

Questions:

Since AggreagateRoot is responsible for saving all of the events to event store, how can it know for existence of all of this sub events that happened in it's tree.
In this transaction we generated 2 events. Should we store them inside event store as array of events, or we should store them simply one event after another. In order for AR to be in consistent state, both events must be applied in single transaction.
In OtherSubEntity we have listen method that is listening to SubEntityEvent. Since on re-hydration from event store we will call SubEntity.when(SubEntityEvent event), how to prevent OtherSubEntity.listen to be called at the same time?
Since aggregate root needs to send 1 resulting message down the pipeline (in my case AMQP message), I suppose that this message will actually be projection of this 2 domain events - basically a read model DTO. Where should I create this model? Something similar as read model just with propagation?

I think this is too much for one question. I will say that typically, my ARs are not responsible for saving events. Instead it simply returns events. Process is like this. The aggregate id is used to load all events from storage. Events and command are passed to handler. Handler rehydrates aggregate from events and calls use case method with necessary dependencies and command data. Events get returned from aggregate, thru handler, and back to infrastructure. Infrastructure persists events. Event listeners are a different matter, not part of aggregate. — Kasey Speakman
– Kasey Speakman, Commented Feb 1, 2017 at 23:17
Is your question in the context of an existing CQRS framework or you try to develop your own framework/library? — Constantin Galbenu
– Constantin Galbenu, Commented Feb 2, 2017 at 0:48

VoiceOfUnreason · Accepted Answer · 2017-02-02 00:07:30Z

Since AggreagateRoot is responsible for saving all of the events to event store, how can it know for existence of all of this sub events that happened in it's tree.

It asks for them. Conceptually, the aggregate root is responsible for all state changes in the aggregate, but it can delegate that work to subordinate entities, which will bubble up the events to the root, which in turn can make them available for storage.

In this transaction we generated 2 events. Should we store them inside event store as array of events, or we should store them simply one event after another. In order for AR to be in consistent state, both events must be applied in single transaction.

Logically, they are a linear history. In practice, your find persistence solutions that treat the List of "new" events to be saved as a single commit. During replay, the events are typically handled one at a time.

Since the AR is loaded directly from a history recorded in the book of record, it has all of the events (or more specifically, any save made from a state that doesn't include all of the events should be rejected), so transient failures to maintain the invariant don't matter, as by the end of the commit the invariant has been re-established.

That said, your event history was written by an old version of the aggregate, but the command is being processed by the current implementation, which may not have the same understanding of "the" invariant.

That aside, in many cases it is possible to change the representation of your events so that the invariant is established after each event.

In OtherSubEntity we have listen method that is listening to SubEntityEvent. Since on re-hydration from event store we will call SubEntity.when(SubEntityEvent event), how to prevent OtherSubEntity.listen to be called at the same time?

There are a couple possibilities -- since re-hyrdration is simply a replay of state, you can defer the listening during replay. You can also work to make sure that the response to the event is idempotent (processing the event twice has the same effect, so it doesn't matter that you "hear" the event before your own replay has finished.

Since aggregate root needs to send 1 resulting message down the pipeline

That constraint needs more context. What message are you thinking of? why does the aggregate need to send it?

WARNING: having answered, let me suggest you abandon this approach. CQRS+ES works much better with an onion architecture, with the model at the core. Coupling the aggregate in your model to your app and your persistence makes the whole exercise much more difficult than it needs to be.

class CommandHandler{
    // the model interface "injected" into your command handler
    Model.API theModel;

    void handle(Command command){
        History priorHistory = repo.getHistoryById(command.arId);
        History updatedHistory = theModel.simulate(priorHistory, command);
        repo.compareAndSwap(priorHistory, updateHistory);
    }
}

Stack Exchange Network

CQRS + Event sourcing - event groups and event propagation

1 Answer 1

Your Answer

Hot Network Questions

CQRS + Event sourcing - event groups and event propagation

1 Answer 1

Your Answer

Sign up or log in

Post as a guest

Related

Hot Network Questions