@Test void testEventsFromArrayMatchingAreForwarded() { EventMessage<String> event1 = GenericEventMessage.asEventMessage("accept"); EventMessage<String> event2 = GenericEventMessage.asEventMessage("fail"); EventMessage<String> event3 = GenericEventMessage.asEventMessage("accept"); testSubject.appendEvents(event1, event2, event3); verify(mockStorage).appendEvents(asList(event1, event3)); }
@Override public void appendEvents(EventMessage<?>... events) { delegate.appendEvents(Arrays.stream(events).filter(filter).collect(Collectors.toList())); }
FilteringEventStorageEngine implements EventStorageEngine { @Override public void appendEvents(EventMessage<?>... events) { delegate.appendEvents(Arrays.stream(events).filter(filter).collect(Collectors.toList())); } }
FilteringEventStorageEngine implements EventStorageEngine { @Override public void appendEvents(EventMessage<?>... events) { delegate.appendEvents(Arrays.stream(events).filter(filter).collect(Collectors.toList())); } FilteringEventStorageEngine(EventStorageEngine delegate, Predicate<? super EventMessage<?>> filter); }
FilteringEventStorageEngine implements EventStorageEngine { @Override public void appendEvents(EventMessage<?>... events) { delegate.appendEvents(Arrays.stream(events).filter(filter).collect(Collectors.toList())); } FilteringEventStorageEngine(EventStorageEngine delegate, Predicate<? super EventMessage<?>> filter); @Override void appendEvents(EventMessage<?>... events); @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); @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); }
FilteringEventStorageEngine implements EventStorageEngine { @Override public void appendEvents(EventMessage<?>... events) { delegate.appendEvents(Arrays.stream(events).filter(filter).collect(Collectors.toList())); } FilteringEventStorageEngine(EventStorageEngine delegate, Predicate<? super EventMessage<?>> filter); @Override void appendEvents(EventMessage<?>... events); @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); @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);  }