Ретроактивный шаблон и как заменить события

Question

Я пытаюсь получить рабочий код, который использует Event Sourcing и ретроактивный шаблон, как описано Мартином Фаулером.

Я немного растерялся из-за того, как события ставятся в очередь?так как события отправляются одно за другим в процесс EventProcessor и функцию реверса.В своей статье он говорит, что класс EventProcessor будет иметь доступ к очереди событий.

, но если мой / его класс EventProcessor будет выглядеть так

public class DomainEventProcessor {

    public void process(DomainEvent domainEvent) {

        if (domainEvent instanceof ReplacementEvent) {
            processReplacement((ReplacementEvent)domainEvent);
        } else {
            domainEvent.process();
        }
    }

    public void reverse(DomainEvent domainEvent) {
        domainEvent.reverse();
    }

    /*
    // replace the  event in event queue
     */
    private void processReplacement(ReplacementEvent e) {


    }
}

Интерфейс событий домена

public abstract class DomainEvent implements Serializable {

    public abstract void process();

    public abstract void reverse();
}

пример реализации события выглядит следующим образом

public class BalanceUpdateEvent extends DomainEvent {

    private String subscriberId;

    private String balanceId;

    private Long delta;

    public String getSubscriberId() {
        return subscriberId;
    }

    public String getBalanceId() {
        return balanceId;
    }

    public Long getDelta() {
        return delta;
    }

    @Override
    public void process() {
    // do nothing
    }

    @Override
    public void reverse() {
        SubscriberDomain subscriberDomain = SubscriberAggregate.getSubscriber(subscriberId);
        subscriberDomain.handleReverseEvent(this);
    }
}

Каждое событие знает, какой объект домена (который имеет состояние + бизнес-логика) может обрабатывать себя для внесения изменений в состояние объекта домена.

Answer 1

public class DomainEventProcessor {

    private EventStore es;

    public DomainEventProcessor(EventStore es) {
        this.es = es;
    }

    public void replay(String aggregateId,Date after) {
        List<DomainEvent> events = es.loadEventStreamAfter(aggregateId,after);
        events.sort(Comparator.comparing(DomainEvent::getTimestamp));
        for(DomainEvent event : events) {
            if (event instanceof ReplacementEvent) {
                ReplacementEvent re = (ReplacementEvent)event;
                re.getReplacement().process();
            } else {
                event.process();
            }
        }
    }

    public void reverse(String aggregateId,Date after) {
        List<DomainEvent> events = es.loadEventStreamAfter(aggregateId,after);
        events.sort(Comparator.comparing(DomainEvent::getTimestamp).reversed());
        for(DomainEvent event : events) {
            if (event instanceof ReplacementEvent) {
                ReplacementEvent re = (ReplacementEvent)event;
                re.getOriginal().reverse();
            } else {
                event.reverse();
            }
        }
    }
}