2

Usually when I read Event Sourcing examples - they are pretty simple. When trying to implement something even more complex - it seems to get pretty ugly pretty quickly.

Let me explain this on one simple example:

I have a Car and car have Door (2 doors only). Now, the single use-case I have here is startCar - which will open the Door and start the car after the door is open. More complex behavior here is that Door is an observer and it listens to the changes of other door and sync it's state here - so If one Door is open, other one will open also.

I start with two Adapters (Ports & Adapters architecture) (one for HTTP - RestController and other for AMQP - AMQPListener) which in the end just map it's input data to common mapped command object (StartCarCommand) inside CarApplicationService which is basically ApplicationService that is responsible for handling use-cases. This first part of the code is not important for the question - just given for the completeness.

class StartCarCommandMapper{
    StartCarCommand map(StartCarRequest request){
        return new StartCarCommand(request.carId, request.doorNumber);
    }
    StartCarCommand map(StartCarMessage message){
        return new StartCarCommand(message.vehicleIdentifier, message.whichDoorToOpen);
    }
}

class RestController{
    //DTO is specified by the sender
    @AllArgsConstructor
    class StartCarRequest{
        int carId;
        int doorNumber;
    }
    private CarApplicationService service;
    private StartCarCommandMapper mapper;
    void startCarRestEndpoint(StartCarRequest request){
        service.startCar(mapper.map(request));
    }
}

class AMQPListener{
    //DTO is specified by the sender
    @AllArgsConstructor
    class StartCarMessage{
        int vehicleIdentifier;
        int whichDoorToOpen;
    }
    private CarApplicationService service;
    private StartCarCommandMapper mapper;
    void onMessage(StartCarMessage message){
        service.startCar(mapper.map(message));
    }
}

class CarRepository{
    CarAggregate getCarById(CarId carId){
        //this would load all events from stream and apply 1 by 1
        return new CarAggregate(carId, Set.of(
                new Door(DoorId.of(1), false),
                new Door(DoorId.of(2), false)
        ));
    }

    public void saveEventStream(CarId carId, List<Object> allEvents) {
        //here we save all events for this aggregate Id to database
    }
}

class CarApplicationService{
    @AllArgsConstructor
    public static class StartCarCommand{
        int carId;
        int doorId;
    }

    private CarRepository repository;

    void startCar(StartCarCommand command){
        CarId carId = CarId.of(command.carId);
        CarAggregate aggregate = repository.getCarById(carId);
        List<DoorOpenedEvent> event1 = aggregate.openDoor(DoorId.of(command.doorId));
        CarStartedEvent event2 = aggregate.startCar();

        List<Object> allEvents = new ArrayList<>(event1);
        allEvents.add(event2);
        repository.saveEventStream(carId, allEvents);
    }
}

Now we come to the interesting part of the domain - Aggregate root which is Car and one entity/vo - Door.

@Builder(toBuilder = true)
class CarAggregate{
    private CarId id;
    private Map<DoorId, Door> doors;
    private boolean started;
    private Set<Door> observers;

    public CarAggregate(CarId id, Set<Door> doors){
        this.id = id;
        this.doors.putAll(doors.stream().collect(Collectors.toMap(Door::getDoorId, Function.identity())));
        this.started = false;
        this.observers.addAll(doors);
    }

    public List<DoorOpenedEvent> openDoor(DoorId doorId) {
        if(!doors.containsKey(doorId)){
            throw new DoorsWithThatIdNotPresent();
        }
        List<DoorOpenedEvent> allEvents = new ArrayList<>();
        DoorOpenedEvent event = doors.get(doorId).open();
        allEvents.add(event);

        //This should be done in loop, since one change can lead to other change etc.
        List<DoorOpenedEvent> fromObservables = observers.stream()
                .map(o -> o.notifyObserver(event))
                .flatMap(Optional::stream)
                .collect(Collectors.toList());

        allEvents.addAll(fromObservables);

        fromObservables.forEach(this::applyEvent); //!!!!<<--- ONLY NOW TO APPLY ALL EVENTS?
        return allEvents;
    }

    public CarStartedEvent startCar() {
        if(started){
            throw new CarAlreadyRunning();
        }
        CarStartedEvent event = CarStartedEvent.of();
        applyEvent(event);
        return event;
    }

    void applyEvent(DoorOpenedEvent event){
        doors.get(DoorId.of(event.getDoorId())).applyEvent(event);
    }

    void applyEvent(CarStartedEvent event){
        started = true;
    }

    class DoorsWithThatIdNotPresent extends RuntimeException{}
    class CarAlreadyRunning extends RuntimeException{}
}

@Builder(toBuilder = true)
@Getter
class Door{
    private DoorId doorId;
    private boolean open;

    public DoorOpenedEvent open(){
        if(open){
            throw new DoorAlreadyOpenException();
        }
        DoorOpenedEvent event = DoorOpenedEvent.of(doorId.getValue());
        //applyEvent(event); //SHOULD IT BE HERE, OR TO BUBBLE UP?
        return event;
    }

    Optional<DoorOpenedEvent> notifyObserver(DoorOpenedEvent event){
        return syncWithOtherDoor(event);
    }

    private Optional<DoorOpenedEvent> syncWithOtherDoor(DoorOpenedEvent event){
        if(!this.open) {    //opposite doors are open now, but we are closed, so sync with other door
            return Optional.of(DoorOpenedEvent.of(doorId.getValue()));
        }
        return Optional.empty();
    }

    void applyEvent(DoorOpenedEvent event){
        open = true;
    }

    class DoorAlreadyOpenException extends RuntimeException{}
}

Questions are:

  1. Should I apply DoorOpenedEvent immediatelly inside Door, or should I bubble it up to the Aggregate root (in this case Car) and only then apply it to the all entities in hierarchy below (as I do now with fromObservables.forEach(this::applyEvent);)?
  2. Is the current Observer pattern correct, since I'm passing only generated Domain Events to interested object parties - but those objects now have to guess what changed only based on those events (since real changes will happen only after when I call fromObservables.forEach(this::applyEvent);?
3

I am going to address several issues I found in your design.

You are leaking events aggregation from your aggregates

How events are managed within an aggregate is responsibility of the aggregate itself. If you take a look at CarApplicationService, your aggregate methods return event(s) which is then processed by application layer and submitted to repository. This can and will lead to unnecessary bugs, when someone forgets to add a recorded aggregate event to the list of events represented by a local variable, which is the passed to the repository.

An aggregate should record its events internally, and provide a read-only mechanism to fetch the recorded events. This is why in event sourced systems, aggregates commonly inherit from an abstract aggregate, which encapsulates the event processing logic:

public abstract class EventSourcedAggregate {
    private final List<Event> recordedEvents = new ArrayList<>();

    public Collection<Event> getRecordedEvents() {
        return this.recordedEvents;
    }

    protected final void applyEvent(Event event) {
        this.recordedEvents.add(event);

        applyRecordedEvent(event);
    }

    protected abstract void applyRecordedEvent(Event event);
}

Your repository interface design is wrong

You are returning a CarAggregate instance from your get method, but you are passing a list of events for saving. Not only is this interface inconsistent, but events are an implementation detail. On top of that, you're currently accepting a List<Object> events argument, anyone can pass a list of anything to the method.

Processing events is an implementation detail of the repository layer, on the application layer level, you're only interested in this process (runtime necessary constructs omitted for simplicity):

class CarApplicationLayerService {

    private CarRepository carRepository;

    public void performOperationOnCar(PersonOperationOnCarCommand command) {
        Car car = carRepository.getById(command.carId);

        car.performOperation(command.someField);

        carRepository.save(car);
    }
}

An aggregate can be event sourced, its parts cannot

In your example you have a CarAggregate which contains a Door - which itself is also event sourced. If you want to make a Door event sourced on its own, then the Door should be a separate aggregate, living along a CarAggregate, not within it.

How do you decide if something is a separate aggregate or a nested entity?

And object should be a nested entity if:

  • all its operations should be performed within a specific context (an executing aggregate, e.g. the CarAggregate provides an executing context for a Door),
  • the specific context introduces constraints on the given entity based on the context's state, while the state does not make sense to live in the entity itself,
  • the object does not need to be necessarily identified on its own and can use the context's identifier as a pointer.

On the other hand, an object should be a separate aggregate if:

  • it is supposed to be event sourced,
  • a possible context introduces zero constraints on the object itself (the nesting is unnecessary, since the context wouldn't protect anything anyway).

The business rule to fulfil is the following one: When a car's door opens, other doors should open as well.

If you find out a Door is in fact a nested entity, your design actually simplifies. Since a door (now being a nested entity) cannot be event sourced, instead of having a DoorOpenedEvent, we will have a CarDoorOpenedEvent. And since all doors should have their state synchronised, the event can be simplified by containing only a reference of a car and completely ignore which door was opened, because in the end all doors should be opened anyway. Therefore your design changes to this:

class CarDoorOpenedEvent implements Event {
    private String carId;

    // getters omitted
}

class Door {
    private long id;
    private boolean open = false;

    public void open() {
        this.open = true;
    }
}

class CarAggregate extends EventSourcedAggregate {
    private String id;
    private Map<Long, Door> doors = new HashMap<>();

    public void openDoor(Long doorId) {
        Door door = doors.get(doorId);
        if (door == null) {
            throw new DoorNotFoundException();
        }

        applyEvent(new CarDoorOpenedEvent(this.id));
    }

    private void applyEvent(CarDoorOpenedEvent event) {
        this.doors.values().forEach(Door::open);
    }
}

But what if you find out that a Door is actually an event sourced aggregate, because you want to track the state of a door at a given time?

If that's the case, then you'll need to introduce a process manager, which will process a command to open a door on a car, by loading all doors belonging to a given car and then sequentially opening one door after another, while the process manager is a state tracker, ensuring that all doors were correctly opened. Because unlike aggregates which accept commands and produce events, process managers accept events and produce commands.


Answers to your questions from comments follow:

[...] If using saga, I guess I would need to artificially convert this to entities with key CarId_DoorId?

Yes, you would need to introduce some kind of an identity so that the door can be referenced.

So lets say in my example - Door is a nested entity - lets say that that OpenDoor command is succeeding only if car has Red color. And when Car checks it color, it can forward command down to Door. Now as you said - I could think of using process manager here and store all doors in an artificial entity CarId_DoorId. But this invariant has to be transactional and consistent - so process manager is not OK here. How would you implement that observer, but without simplification?

The design does not change too much from the original implementation I've suggested in the earlier part of this answer. If I understand your rule correctly, the implementation would look like this:

enum Color {
    BLACK,
    RED
}

class CarAggregate extends EventSourcedAggregate {
    private String id;
    private Color color;
    private Map<Long, Door> doors = new HashMap<>();

    public void openDoor(Long doorId) {
        if (this.color != Color.RED) {
            throw new InvalidCarColorException();
        }

        Door door = doors.get(doorId);
        if (door == null) {
            throw new DoorNotFoundException();
        }

        applyEvent(new CarDoorOpenedEvent(this.id));
    }

    private void applyEvent(CarDoorOpenedEvent event) {
        this.doors.values().forEach(Door::open);
    }
}
  • Thanks for thorough answer. Related to 1st one - I guess both approaches are OK. I do not see that big problem with events returned from Aggregate, since all errors should be handled by tests anyway. Related to 2nd - I agree with you (this part was just here for completeness and implemented ad-hoc so it has those problems you mentioned). Now the most important - in your implementation you simplified my design - but the example I have is artificial one, just to be able to ask for these implementation problems when I have nested Objects as in this case. – Bojan Vukasovic Nov 23 '20 at 9:40
  • Using saga/proc. manager is an option of course, but I guess this would complicate design a lot (and not to mention eventual consistency). Door in this case is basically Value Object(s) and as such has no identity outside Car Entity. If using saga, I guess I would need to artificially convert this to entities with key CarId_DoorId? – Bojan Vukasovic Nov 23 '20 at 9:42
  • So lets say in my example - Door is a nested entity - lets say that that OpenDoor command is succeeding only if car has Red color. And when Car checks it color, it can forward command down to Door. Now as you said - I could think of using process manager here and store all doors in an artificial entity CarId_DoorId. But this invariant has to be transactional and consistent - so process manager is not OK here. How would you implement that observer, but without simplification? – Bojan Vukasovic Nov 23 '20 at 9:49
  • When I said returning events from Aggregate - In this case, repository would only be there to save Events, and load Events. Some AggregateFactory would be in charge of constructing Aggregate from events (basically same thing that repo is doing). Not saying that this is in any case better than standard approach, just that I guess it is one more option. – Bojan Vukasovic Nov 23 '20 at 9:57
  • @BojanVukasovic I have edited the original answer to provide answers to your questions from the comments. – Andy Nov 23 '20 at 10:15

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.