@Test public void testPeekOnNonEmptyStreamShouldContainSomeElement() throws InterruptedException { GenericTrackedDomainEventMessage<String> firstMessage = trackedDomainEvent("foo"); KafkaMessageStream testSubject = stream(Arrays.asList(firstMessage, trackedDomainEvent("bar"))); assertThat(testSubject.peek().isPresent()).isTrue(); assertThat(testSubject.peek().get()).isEqualTo(firstMessage); }
@Override public Optional<TrackedEventMessage<?>> peek() { return Optional.ofNullable( peekedEvent == null && !hasNextAvailable(0, TimeUnit.NANOSECONDS) ? null : peekedEvent.value() ); }
KafkaMessageStream implements TrackingEventStream { @Override public Optional<TrackedEventMessage<?>> peek() { return Optional.ofNullable( peekedEvent == null && !hasNextAvailable(0, TimeUnit.NANOSECONDS) ? null : peekedEvent.value() ); } }
KafkaMessageStream implements TrackingEventStream { @Override public Optional<TrackedEventMessage<?>> peek() { return Optional.ofNullable( peekedEvent == null && !hasNextAvailable(0, TimeUnit.NANOSECONDS) ? null : peekedEvent.value() ); } @SuppressWarnings("WeakerAccess")  KafkaMessageStream(Buffer<KafkaEventMessage> buffer, Registration closeHandler); }
KafkaMessageStream implements TrackingEventStream { @Override public Optional<TrackedEventMessage<?>> peek() { return Optional.ofNullable( peekedEvent == null && !hasNextAvailable(0, TimeUnit.NANOSECONDS) ? null : peekedEvent.value() ); } @SuppressWarnings("WeakerAccess")  KafkaMessageStream(Buffer<KafkaEventMessage> buffer, Registration closeHandler); @Override Optional<TrackedEventMessage<?>> peek(); @Override boolean hasNextAvailable(int timeout, TimeUnit unit); @Override TrackedEventMessage<?> nextAvailable(); @Override void close(); }
KafkaMessageStream implements TrackingEventStream { @Override public Optional<TrackedEventMessage<?>> peek() { return Optional.ofNullable( peekedEvent == null && !hasNextAvailable(0, TimeUnit.NANOSECONDS) ? null : peekedEvent.value() ); } @SuppressWarnings("WeakerAccess")  KafkaMessageStream(Buffer<KafkaEventMessage> buffer, Registration closeHandler); @Override Optional<TrackedEventMessage<?>> peek(); @Override boolean hasNextAvailable(int timeout, TimeUnit unit); @Override TrackedEventMessage<?> nextAvailable(); @Override void close();  }