@Test void createTailToken() { EventMessage pubToStreamA = GenericEventMessage.asEventMessage("Event1"); eventStoreA.publish(pubToStreamA); EventMessage pubToStreamB = GenericEventMessage.asEventMessage("Event2"); eventStoreB.publish(pubToStreamB); MultiSourceTrackingToken tailToken = testSubject.createTailToken(); assertEquals(-1L, tailToken.getTokenForStream("eventStoreA").position().getAsLong()); assertEquals(-1L, tailToken.getTokenForStream("eventStoreB").position().getAsLong()); }
@Override public MultiSourceTrackingToken createTailToken() { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTailToken())); return new MultiSourceTrackingToken(tokenMap); }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceTrackingToken createTailToken() { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTailToken())); return new MultiSourceTrackingToken(tokenMap); } }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceTrackingToken createTailToken() { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTailToken())); return new MultiSourceTrackingToken(tokenMap); } protected  MultiStreamableMessageSource(Builder builder); }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceTrackingToken createTailToken() { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTailToken())); 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 createTailToken() { Map<String, TrackingToken> tokenMap = new HashMap<>(); eventStreams.forEach(streamableMessageSource -> tokenMap.put(streamableMessageSource.sourceId(), streamableMessageSource.createTailToken())); 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);  }