@Test public void returnsResultOfProjection() { TestEvent testEvent1 = new TestEvent("test1"); TestEvent testEvent2 = new TestEvent("test2"); ImmutableList<EventRecord<TestEvent>> events = ImmutableList.of( new EventRecord<>( AGGREGATE_ID_1, 3, 1, "test1", UUID.randomUUID(), Instant.now(), ImmutableMap.of(), testEvent1), new EventRecord<>( AGGREGATE_ID_1, 4, 1, "test1", UUID.randomUUID(), Instant.now(), ImmutableMap.of(), testEvent2)); EventReadResult repositoryResult = new EventReadResult<>(events, 0, 1, 2, true); List<TestEvent> expectedEvents = ImmutableList.of(testEvent1, testEvent2); TestState expectedState = new TestState("state1"); when(eventRepository.read(any(), anyInt(), anyInt())).thenReturn(repositoryResult); when(aggregateProjection.apply(any(), any(), (Iterable) any())).thenReturn(expectedState); ImmutableAggregate<TestState, TestEvent> result = repository.load(AGGREGATE_ID_1); verify(eventRepository).read(eq(AGGREGATE_ID_1), eq(0), anyInt()); ArgumentCaptor<Iterable<TestEvent>> passedEvents = ArgumentCaptor.forClass((Class) Iterable.class); verify(aggregateProjection).empty(); verify(aggregateProjection).apply( eq(AGGREGATE_ID_1), eq(new TestState("empty")), passedEvents.capture()); Assert.assertThat( passedEvents.getValue(), org.hamcrest.Matchers.contains(expectedEvents.toArray())); verifyNoMoreInteractions(eventRepository, aggregateProjection); Assert.assertEquals(expectedState, result.state()); }
@Override public ImmutableAggregate<TState, TEvent> load(final String aggregateId) { TState state = projection.empty(); int currentStreamPosition = 0; try { while (true) { logger.debug("Reading events for {} from {}", aggregateId, currentStreamPosition); EventReadResult<TEvent> readResult = eventRepository.read( aggregateId, currentStreamPosition, maxEventsPerRead); if (readResult == null || readResult.getEvents().isEmpty()) { return DefaultImmutableAggregate.createNew( projection, aggregateId); } Iterable<TEvent> events = readResult .getEvents() .stream() .map(EventRecord::getEvent) .collect(Collectors.toList()); state = projection.apply(aggregateId, state, events); if (readResult.isEndOfStream()) { return DefaultImmutableAggregate.fromExisting( projection, aggregateId, readResult.getLastVersion(), state); } currentStreamPosition = readResult.getNextVersion(); } } finally { if (currentStreamPosition > LARGE_EVENT_STREAM_WARNING_CUTOFF) { logger.warn( "Read large stream {} consisting of {} events - consider snapshotting?", aggregateId, currentStreamPosition); } } }
DefaultAggregateRepository implements AggregateRepository<TState, TEvent> { @Override public ImmutableAggregate<TState, TEvent> load(final String aggregateId) { TState state = projection.empty(); int currentStreamPosition = 0; try { while (true) { logger.debug("Reading events for {} from {}", aggregateId, currentStreamPosition); EventReadResult<TEvent> readResult = eventRepository.read( aggregateId, currentStreamPosition, maxEventsPerRead); if (readResult == null || readResult.getEvents().isEmpty()) { return DefaultImmutableAggregate.createNew( projection, aggregateId); } Iterable<TEvent> events = readResult .getEvents() .stream() .map(EventRecord::getEvent) .collect(Collectors.toList()); state = projection.apply(aggregateId, state, events); if (readResult.isEndOfStream()) { return DefaultImmutableAggregate.fromExisting( projection, aggregateId, readResult.getLastVersion(), state); } currentStreamPosition = readResult.getNextVersion(); } } finally { if (currentStreamPosition > LARGE_EVENT_STREAM_WARNING_CUTOFF) { logger.warn( "Read large stream {} consisting of {} events - consider snapshotting?", aggregateId, currentStreamPosition); } } } }
DefaultAggregateRepository implements AggregateRepository<TState, TEvent> { @Override public ImmutableAggregate<TState, TEvent> load(final String aggregateId) { TState state = projection.empty(); int currentStreamPosition = 0; try { while (true) { logger.debug("Reading events for {} from {}", aggregateId, currentStreamPosition); EventReadResult<TEvent> readResult = eventRepository.read( aggregateId, currentStreamPosition, maxEventsPerRead); if (readResult == null || readResult.getEvents().isEmpty()) { return DefaultImmutableAggregate.createNew( projection, aggregateId); } Iterable<TEvent> events = readResult .getEvents() .stream() .map(EventRecord::getEvent) .collect(Collectors.toList()); state = projection.apply(aggregateId, state, events); if (readResult.isEndOfStream()) { return DefaultImmutableAggregate.fromExisting( projection, aggregateId, readResult.getLastVersion(), state); } currentStreamPosition = readResult.getNextVersion(); } } finally { if (currentStreamPosition > LARGE_EVENT_STREAM_WARNING_CUTOFF) { logger.warn( "Read large stream {} consisting of {} events - consider snapshotting?", aggregateId, currentStreamPosition); } } } DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final int maxEventsPerRead);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final Function<? super TEvent, String> typeResolver);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final Function<? super TEvent, String> typeResolver,
            final int maxEventsPerRead); }
DefaultAggregateRepository implements AggregateRepository<TState, TEvent> { @Override public ImmutableAggregate<TState, TEvent> load(final String aggregateId) { TState state = projection.empty(); int currentStreamPosition = 0; try { while (true) { logger.debug("Reading events for {} from {}", aggregateId, currentStreamPosition); EventReadResult<TEvent> readResult = eventRepository.read( aggregateId, currentStreamPosition, maxEventsPerRead); if (readResult == null || readResult.getEvents().isEmpty()) { return DefaultImmutableAggregate.createNew( projection, aggregateId); } Iterable<TEvent> events = readResult .getEvents() .stream() .map(EventRecord::getEvent) .collect(Collectors.toList()); state = projection.apply(aggregateId, state, events); if (readResult.isEndOfStream()) { return DefaultImmutableAggregate.fromExisting( projection, aggregateId, readResult.getLastVersion(), state); } currentStreamPosition = readResult.getNextVersion(); } } finally { if (currentStreamPosition > LARGE_EVENT_STREAM_WARNING_CUTOFF) { logger.warn( "Read large stream {} consisting of {} events - consider snapshotting?", aggregateId, currentStreamPosition); } } } DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final int maxEventsPerRead);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final Function<? super TEvent, String> typeResolver);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final Function<? super TEvent, String> typeResolver,
            final int maxEventsPerRead); @Override ImmutableAggregate<TState, TEvent> load(final String aggregateId); @Override int append(
            final String aggregateId,
            final Iterable<? extends TEvent> events,
            final ExpectedVersion expectedVersion,
            final Map<String, String> metadata); @Override ImmutableAggregate<TState, TEvent> save(
            @NotNull final Aggregate<TState, TEvent> aggregate,
            final boolean atomic,
            final Map<String, String> metadata); }
DefaultAggregateRepository implements AggregateRepository<TState, TEvent> { @Override public ImmutableAggregate<TState, TEvent> load(final String aggregateId) { TState state = projection.empty(); int currentStreamPosition = 0; try { while (true) { logger.debug("Reading events for {} from {}", aggregateId, currentStreamPosition); EventReadResult<TEvent> readResult = eventRepository.read( aggregateId, currentStreamPosition, maxEventsPerRead); if (readResult == null || readResult.getEvents().isEmpty()) { return DefaultImmutableAggregate.createNew( projection, aggregateId); } Iterable<TEvent> events = readResult .getEvents() .stream() .map(EventRecord::getEvent) .collect(Collectors.toList()); state = projection.apply(aggregateId, state, events); if (readResult.isEndOfStream()) { return DefaultImmutableAggregate.fromExisting( projection, aggregateId, readResult.getLastVersion(), state); } currentStreamPosition = readResult.getNextVersion(); } } finally { if (currentStreamPosition > LARGE_EVENT_STREAM_WARNING_CUTOFF) { logger.warn( "Read large stream {} consisting of {} events - consider snapshotting?", aggregateId, currentStreamPosition); } } } DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final int maxEventsPerRead);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final Function<? super TEvent, String> typeResolver);  DefaultAggregateRepository(
            final EventRepository<TEvent> eventRepository,
            final AggregateProjection<TState, TEvent> projection,
            final Function<? super TEvent, String> typeResolver,
            final int maxEventsPerRead); @Override ImmutableAggregate<TState, TEvent> load(final String aggregateId); @Override int append(
            final String aggregateId,
            final Iterable<? extends TEvent> events,
            final ExpectedVersion expectedVersion,
            final Map<String, String> metadata); @Override ImmutableAggregate<TState, TEvent> save(
            @NotNull final Aggregate<TState, TEvent> aggregate,
            final boolean atomic,
            final Map<String, String> metadata);  }