@Test void testStagedEventsNotDuplicatedAfterCommit() { List<DomainEventMessage<?>> events = createEvents(10); testSubject.publish(events.subList(0, 2)); DefaultUnitOfWork<Message<?>> outerUoW = DefaultUnitOfWork.startAndGet(null); testSubject.publish(events.subList(2, 4)); DefaultUnitOfWork<Message<?>> innerUoW = DefaultUnitOfWork.startAndGet(null); testSubject.publish(events.subList(4, events.size())); Consumer<UnitOfWork<Message<?>>> assertCorrectEventCount = uow -> { assertEquals(10, testSubject.readEvents(AGGREGATE).asStream().count()); }; innerUoW.onPrepareCommit(assertCorrectEventCount); innerUoW.afterCommit(assertCorrectEventCount); innerUoW.onCommit(assertCorrectEventCount); outerUoW.onPrepareCommit(assertCorrectEventCount); outerUoW.afterCommit(assertCorrectEventCount); outerUoW.onCommit(assertCorrectEventCount); innerUoW.commit(); outerUoW.commit(); }
@Override protected void afterCommit(List<? extends EventMessage<?>> events) { producer.fetchIfWaiting(); }
EmbeddedEventStore extends AbstractEventStore { @Override protected void afterCommit(List<? extends EventMessage<?>> events) { producer.fetchIfWaiting(); } }
EmbeddedEventStore extends AbstractEventStore { @Override protected void afterCommit(List<? extends EventMessage<?>> events) { producer.fetchIfWaiting(); } protected  EmbeddedEventStore(Builder builder); }
EmbeddedEventStore extends AbstractEventStore { @Override protected void afterCommit(List<? extends EventMessage<?>> events) { producer.fetchIfWaiting(); } protected  EmbeddedEventStore(Builder builder); static Builder builder(); @PreDestroy void shutDown(); @Override TrackingEventStream openStream(TrackingToken trackingToken); }
EmbeddedEventStore extends AbstractEventStore { @Override protected void afterCommit(List<? extends EventMessage<?>> events) { producer.fetchIfWaiting(); } protected  EmbeddedEventStore(Builder builder); static Builder builder(); @PreDestroy void shutDown(); @Override TrackingEventStream openStream(TrackingToken trackingToken);  }