@Test void testAggregateEventsAreReadFromActiveWhenNoHistoricEventsAvailable() { DomainEventMessage<String> event1 = new GenericDomainEventMessage<>("type", "aggregate", 0, "test1"); DomainEventMessage<String> event2 = new GenericDomainEventMessage<>("type", "aggregate", 1, "test2"); when(historicStorage.readEvents(eq("aggregate"), anyLong())).thenReturn(DomainEventStream.empty()); when(activeStorage.readEvents(eq("aggregate"), anyLong())).thenReturn(DomainEventStream.of(event1, event2)); DomainEventStream actual = testSubject.readEvents("aggregate", 0); verify(activeStorage, never()).readEvents(anyString(), anyLong()); assertSame(event1, actual.peek()); assertSame(event1, actual.next()); assertSame(event2, actual.peek()); assertSame(event2, actual.next()); Long seq = actual.getLastSequenceNumber(); assertNotNull(seq); InOrder inOrder = Mockito.inOrder(historicStorage, activeStorage); inOrder.verify(historicStorage).readEvents("aggregate", 0); inOrder.verify(activeStorage).readEvents("aggregate", 0); assertFalse(actual.hasNext()); }
@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);  }