@Test void testPeekingLastMessageKeepsItAvailable() throws InterruptedException { EventMessage<?> publishedEvent1 = GenericEventMessage.asEventMessage("Event1"); eventStoreA.publish(publishedEvent1); BlockingStream<TrackedEventMessage<?>> stream = testSubject.openStream(null); assertEquals("Event1", stream.peek().map(Message::getPayload).map(Object::toString).orElse("None")); assertTrue(stream.hasNextAvailable()); assertTrue(stream.hasNextAvailable(10, TimeUnit.SECONDS)); }
@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);  }