@Test void testLoadAndSaveAggregate() { String identifier = UUID.randomUUID().toString(); DomainEventMessage event1 = new GenericDomainEventMessage<>("type", identifier, (long) 1, "Mock contents", emptyInstance()); DomainEventMessage event2 = new GenericDomainEventMessage<>("type", identifier, (long) 2, "Mock contents", emptyInstance()); when(mockEventStore.readEvents(identifier)).thenReturn(DomainEventStream.of(event1, event2)); Aggregate<TestAggregate> aggregate = testSubject.load(identifier, null); assertEquals(2, aggregate.invoke(TestAggregate::getHandledEvents).size()); assertSame(event1, aggregate.invoke(TestAggregate::getHandledEvents).get(0)); assertSame(event2, aggregate.invoke(TestAggregate::getHandledEvents).get(1)); assertEquals(0, aggregate.invoke(TestAggregate::getLiveEvents).size()); StubDomainEvent event3 = new StubDomainEvent(); aggregate.execute(r -> r.apply(event3)); CurrentUnitOfWork.commit(); verify(mockEventStore, times(1)).publish((EventMessage) any()); assertEquals(1, aggregate.invoke(TestAggregate::getLiveEvents).size()); assertSame(event3, aggregate.invoke(TestAggregate::getLiveEvents).get(0).getPayload()); }
protected DomainEventStream readEvents(String aggregateIdentifier) { DomainEventStream fullStream = eventStore.readEvents(aggregateIdentifier); return eventStreamFilter != null ? fullStream.filter(eventStreamFilter) : fullStream; }
EventSourcingRepository extends LockingRepository<T, EventSourcedAggregate<T>> { protected DomainEventStream readEvents(String aggregateIdentifier) { DomainEventStream fullStream = eventStore.readEvents(aggregateIdentifier); return eventStreamFilter != null ? fullStream.filter(eventStreamFilter) : fullStream; } }
EventSourcingRepository extends LockingRepository<T, EventSourcedAggregate<T>> { protected DomainEventStream readEvents(String aggregateIdentifier) { DomainEventStream fullStream = eventStore.readEvents(aggregateIdentifier); return eventStreamFilter != null ? fullStream.filter(eventStreamFilter) : fullStream; } protected  EventSourcingRepository(Builder<T> builder); }
EventSourcingRepository extends LockingRepository<T, EventSourcedAggregate<T>> { protected DomainEventStream readEvents(String aggregateIdentifier) { DomainEventStream fullStream = eventStore.readEvents(aggregateIdentifier); return eventStreamFilter != null ? fullStream.filter(eventStreamFilter) : fullStream; } protected  EventSourcingRepository(Builder<T> builder); static Builder<T> builder(Class<T> aggregateType); AggregateFactory<T> getAggregateFactory(); }
EventSourcingRepository extends LockingRepository<T, EventSourcedAggregate<T>> { protected DomainEventStream readEvents(String aggregateIdentifier) { DomainEventStream fullStream = eventStore.readEvents(aggregateIdentifier); return eventStreamFilter != null ? fullStream.filter(eventStreamFilter) : fullStream; } protected  EventSourcingRepository(Builder<T> builder); static Builder<T> builder(Class<T> aggregateType); AggregateFactory<T> getAggregateFactory();  }