@Test public void testNextAvailableOnAProgressiveStreamShouldContainElementInCorrectOrder() throws InterruptedException { GenericTrackedDomainEventMessage<String> firstMessage = trackedDomainEvent("foo"); GenericTrackedDomainEventMessage<String> secondMessage = trackedDomainEvent("bar"); KafkaMessageStream testSubject = stream(Arrays.asList(firstMessage, secondMessage)); assertThat(testSubject.nextAvailable()).isEqualTo(firstMessage); assertThat(testSubject.nextAvailable()).isEqualTo(secondMessage); }
@Override public TrackedEventMessage<?> nextAvailable() { try { return peekedEvent == null ? buffer.take().value() : peekedEvent.value(); } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return null; } finally { peekedEvent = null; } }
KafkaMessageStream implements TrackingEventStream { @Override public TrackedEventMessage<?> nextAvailable() { try { return peekedEvent == null ? buffer.take().value() : peekedEvent.value(); } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return null; } finally { peekedEvent = null; } } }
KafkaMessageStream implements TrackingEventStream { @Override public TrackedEventMessage<?> nextAvailable() { try { return peekedEvent == null ? buffer.take().value() : peekedEvent.value(); } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return null; } finally { peekedEvent = null; } } @SuppressWarnings("WeakerAccess")  KafkaMessageStream(Buffer<KafkaEventMessage> buffer, Registration closeHandler); }
KafkaMessageStream implements TrackingEventStream { @Override public TrackedEventMessage<?> nextAvailable() { try { return peekedEvent == null ? buffer.take().value() : peekedEvent.value(); } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return null; } finally { peekedEvent = null; } } @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 TrackedEventMessage<?> nextAvailable() { try { return peekedEvent == null ? buffer.take().value() : peekedEvent.value(); } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return null; } finally { peekedEvent = null; } } @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();  }