@Test void createTokenSince() throws InterruptedException { EventMessage pubToStreamA = GenericEventMessage.asEventMessage("Event1"); eventStoreA.publish(pubToStreamA); eventStoreA.publish(pubToStreamA); Thread.sleep(20); EventMessage pubToStreamB = GenericEventMessage.asEventMessage("Event2"); eventStoreB.publish(pubToStreamB); MultiSourceTrackingToken createdSinceToken = testSubject.createTokenSince(Duration.ofMillis(10)); assertNull(createdSinceToken.getTokenForStream("eventStoreA")); assertEquals(-1L, createdSinceToken.getTokenForStream("eventStoreB").position().getAsLong()); }
@Override public MultiSourceTrackingToken createTokenSince(Duration duration) { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTokenSince(duration))); return new MultiSourceTrackingToken(tokenMap); }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceTrackingToken createTokenSince(Duration duration) { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTokenSince(duration))); return new MultiSourceTrackingToken(tokenMap); } }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceTrackingToken createTokenSince(Duration duration) { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTokenSince(duration))); return new MultiSourceTrackingToken(tokenMap); } protected  MultiStreamableMessageSource(Builder builder); }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceTrackingToken createTokenSince(Duration duration) { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTokenSince(duration))); return new MultiSourceTrackingToken(tokenMap); } protected  MultiStreamableMessageSource(Builder builder); static Builder builder(); @Override MultiSourceBlockingStream openStream(TrackingToken trackingToken); @Override MultiSourceTrackingToken createTailToken(); @Override MultiSourceTrackingToken createHeadToken(); @Override MultiSourceTrackingToken createTokenAt(Instant dateTime); @Override MultiSourceTrackingToken createTokenSince(Duration duration); }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceTrackingToken createTokenSince(Duration duration) { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTokenSince(duration))); return new MultiSourceTrackingToken(tokenMap); } protected  MultiStreamableMessageSource(Builder builder); static Builder builder(); @Override MultiSourceBlockingStream openStream(TrackingToken trackingToken); @Override MultiSourceTrackingToken createTailToken(); @Override MultiSourceTrackingToken createHeadToken(); @Override MultiSourceTrackingToken createTokenAt(Instant dateTime); @Override MultiSourceTrackingToken createTokenSince(Duration duration);  }