@Test public void testHasNextOnAnInterruptedStreamShouldThrowAnException() throws InterruptedException { try { KafkaMessageStream testSubject = stream(singletonList(trackedDomainEvent("foo"))); Thread.currentThread().interrupt(); assertThat(testSubject.hasNextAvailable(1, DEFAULT_TIMEOUT_UNIT)).isFalse(); } finally { Thread.interrupted(); } }
@Override public boolean hasNextAvailable(int timeout, TimeUnit unit) { try { return peekedEvent != null || (peekedEvent = buffer.poll(timeout, unit)) != null; } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } }
KafkaMessageStream implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit unit) { try { return peekedEvent != null || (peekedEvent = buffer.poll(timeout, unit)) != null; } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } }
KafkaMessageStream implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit unit) { try { return peekedEvent != null || (peekedEvent = buffer.poll(timeout, unit)) != null; } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } @SuppressWarnings("WeakerAccess")  KafkaMessageStream(Buffer<KafkaEventMessage> buffer, Registration closeHandler); }
KafkaMessageStream implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit unit) { try { return peekedEvent != null || (peekedEvent = buffer.poll(timeout, unit)) != null; } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } @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 boolean hasNextAvailable(int timeout, TimeUnit unit) { try { return peekedEvent != null || (peekedEvent = buffer.poll(timeout, unit)) != null; } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } @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();  }