@Test void openStreamWithNullTokenReturnsFirstEvent() throws InterruptedException { EventMessage<Object> message = GenericEventMessage.asEventMessage("Event1"); eventStoreA.publish(message); BlockingStream<TrackedEventMessage<?>> actual = testSubject.openStream(null); assertNotNull(actual); TrackedEventMessage<?> trackedEventMessage = actual.nextAvailable(); assertEquals(message.getIdentifier(), trackedEventMessage.getIdentifier()); assertEquals(message.getPayload(), trackedEventMessage.getPayload()); }
@Override public MultiSourceBlockingStream openStream(TrackingToken trackingToken) { if (trackingToken == null) { return openStream(createTailToken()); } else if (trackingToken instanceof MultiSourceTrackingToken) { return new MultiSourceBlockingStream(eventStreams, (MultiSourceTrackingToken) trackingToken, trackedEventComparator); } throw new IllegalArgumentException("Incompatible token type provided."); }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceBlockingStream openStream(TrackingToken trackingToken) { if (trackingToken == null) { return openStream(createTailToken()); } else if (trackingToken instanceof MultiSourceTrackingToken) { return new MultiSourceBlockingStream(eventStreams, (MultiSourceTrackingToken) trackingToken, trackedEventComparator); } throw new IllegalArgumentException("Incompatible token type provided."); } }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceBlockingStream openStream(TrackingToken trackingToken) { if (trackingToken == null) { return openStream(createTailToken()); } else if (trackingToken instanceof MultiSourceTrackingToken) { return new MultiSourceBlockingStream(eventStreams, (MultiSourceTrackingToken) trackingToken, trackedEventComparator); } throw new IllegalArgumentException("Incompatible token type provided."); } protected  MultiStreamableMessageSource(Builder builder); }
MultiStreamableMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public MultiSourceBlockingStream openStream(TrackingToken trackingToken) { if (trackingToken == null) { return openStream(createTailToken()); } else if (trackingToken instanceof MultiSourceTrackingToken) { return new MultiSourceBlockingStream(eventStreams, (MultiSourceTrackingToken) trackingToken, trackedEventComparator); } throw new IllegalArgumentException("Incompatible token type provided."); } 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 MultiSourceBlockingStream openStream(TrackingToken trackingToken) { if (trackingToken == null) { return openStream(createTailToken()); } else if (trackingToken instanceof MultiSourceTrackingToken) { return new MultiSourceBlockingStream(eventStreams, (MultiSourceTrackingToken) trackingToken, trackedEventComparator); } throw new IllegalArgumentException("Incompatible token type provided."); } 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);  }