@Test void testEventStreamedFromHistoricThenActive() { DomainEventMessage<String> event1 = new GenericDomainEventMessage<>("type", "aggregate", 0, "test1"); DomainEventMessage<String> event2 = new GenericDomainEventMessage<>("type", "aggregate", 1, "test2"); TrackingToken token1 = new GlobalSequenceTrackingToken(1); TrackingToken token2 = new GlobalSequenceTrackingToken(2); TrackedEventMessage<?> trackedEvent1 = new GenericTrackedDomainEventMessage<>(token1, event1); TrackedEventMessage<?> trackedEvent2 = new GenericTrackedDomainEventMessage<>(token2, event2); doReturn(Stream.of(trackedEvent1)).when(historicStorage).readEvents(any(TrackingToken.class), anyBoolean()); doReturn(Stream.of(trackedEvent2)).when(activeStorage).readEvents(any(TrackingToken.class), anyBoolean()); GlobalSequenceTrackingToken startToken = new GlobalSequenceTrackingToken(0); Stream<? extends TrackedEventMessage<?>> actual = testSubject.readEvents(startToken, true); List<? extends TrackedEventMessage<?>> actualList = actual.collect(toList()); assertEquals(2, actualList.size()); assertEquals(Arrays.asList(trackedEvent1, trackedEvent2), actualList); verify(historicStorage).readEvents(startToken, true); verify(activeStorage).readEvents(token1, true); }
@Override public Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock) { Spliterator<? extends TrackedEventMessage<?>> historicSpliterator = historicStorage.readEvents(trackingToken, mayBlock).spliterator(); Spliterator<? extends TrackedEventMessage<?>> merged = new ConcatenatingSpliterator( trackingToken, historicSpliterator, mayBlock, token -> activeStorage.readEvents(token, mayBlock).spliterator() ); return StreamSupport.stream(merged, false); }
SequenceEventStorageEngine implements EventStorageEngine { @Override public Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock) { Spliterator<? extends TrackedEventMessage<?>> historicSpliterator = historicStorage.readEvents(trackingToken, mayBlock).spliterator(); Spliterator<? extends TrackedEventMessage<?>> merged = new ConcatenatingSpliterator( trackingToken, historicSpliterator, mayBlock, token -> activeStorage.readEvents(token, mayBlock).spliterator() ); return StreamSupport.stream(merged, false); } }
SequenceEventStorageEngine implements EventStorageEngine { @Override public Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock) { Spliterator<? extends TrackedEventMessage<?>> historicSpliterator = historicStorage.readEvents(trackingToken, mayBlock).spliterator(); Spliterator<? extends TrackedEventMessage<?>> merged = new ConcatenatingSpliterator( trackingToken, historicSpliterator, mayBlock, token -> activeStorage.readEvents(token, mayBlock).spliterator() ); return StreamSupport.stream(merged, false); } SequenceEventStorageEngine(EventStorageEngine historicStorage, EventStorageEngine activeStorage); }
SequenceEventStorageEngine implements EventStorageEngine { @Override public Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock) { Spliterator<? extends TrackedEventMessage<?>> historicSpliterator = historicStorage.readEvents(trackingToken, mayBlock).spliterator(); Spliterator<? extends TrackedEventMessage<?>> merged = new ConcatenatingSpliterator( trackingToken, historicSpliterator, mayBlock, token -> activeStorage.readEvents(token, mayBlock).spliterator() ); return StreamSupport.stream(merged, false); } SequenceEventStorageEngine(EventStorageEngine historicStorage, EventStorageEngine activeStorage); @Override void appendEvents(List<? extends EventMessage<?>> events); @Override void storeSnapshot(DomainEventMessage<?> snapshot); @Override Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock); @Override DomainEventStream readEvents(String aggregateIdentifier, long firstSequenceNumber); @Override Optional<DomainEventMessage<?>> readSnapshot(String aggregateIdentifier); @Override Optional<Long> lastSequenceNumberFor(String aggregateIdentifier); @Override TrackingToken createTailToken(); @Override TrackingToken createHeadToken(); @Override TrackingToken createTokenAt(Instant dateTime); }
SequenceEventStorageEngine implements EventStorageEngine { @Override public Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock) { Spliterator<? extends TrackedEventMessage<?>> historicSpliterator = historicStorage.readEvents(trackingToken, mayBlock).spliterator(); Spliterator<? extends TrackedEventMessage<?>> merged = new ConcatenatingSpliterator( trackingToken, historicSpliterator, mayBlock, token -> activeStorage.readEvents(token, mayBlock).spliterator() ); return StreamSupport.stream(merged, false); } SequenceEventStorageEngine(EventStorageEngine historicStorage, EventStorageEngine activeStorage); @Override void appendEvents(List<? extends EventMessage<?>> events); @Override void storeSnapshot(DomainEventMessage<?> snapshot); @Override Stream<? extends TrackedEventMessage<?>> readEvents(TrackingToken trackingToken, boolean mayBlock); @Override DomainEventStream readEvents(String aggregateIdentifier, long firstSequenceNumber); @Override Optional<DomainEventMessage<?>> readSnapshot(String aggregateIdentifier); @Override Optional<Long> lastSequenceNumberFor(String aggregateIdentifier); @Override TrackingToken createTailToken(); @Override TrackingToken createHeadToken(); @Override TrackingToken createTokenAt(Instant dateTime);  }