@Test void testHasNextAvailableThrowsAxonServerExceptionWhenStreamFailed() { eventStream.onError(new TestException()); assertThrows(AxonServerException.class, () -> testSubject.hasNextAvailable(0, TimeUnit.SECONDS)); assertThrows(AxonServerException.class, () -> testSubject.hasNextAvailable(0, TimeUnit.SECONDS)); }
@Override public boolean hasNextAvailable(int timeout, TimeUnit timeUnit) { long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { while (peekNullable() == null && System.currentTimeMillis() < deadline) { lock.lock(); try { long waitTime = deadline - System.currentTimeMillis(); if (peekNullable() == null && waitTime > 0) { boolean await = dataAvailable.await(Math.min(waitTime, MAX_AWAIT_AVAILABLE_DATA), TimeUnit.MILLISECONDS); logger.trace(await ? "Signaled new events are available" : "No signal received for new events, exiting await"); } } finally { lock.unlock(); } } return peekNullable() != null; } catch (InterruptedException e) { logger.warn("Event consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } }
EventBuffer implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit timeUnit) { long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { while (peekNullable() == null && System.currentTimeMillis() < deadline) { lock.lock(); try { long waitTime = deadline - System.currentTimeMillis(); if (peekNullable() == null && waitTime > 0) { boolean await = dataAvailable.await(Math.min(waitTime, MAX_AWAIT_AVAILABLE_DATA), TimeUnit.MILLISECONDS); logger.trace(await ? "Signaled new events are available" : "No signal received for new events, exiting await"); } } finally { lock.unlock(); } } return peekNullable() != null; } catch (InterruptedException e) { logger.warn("Event consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } }
EventBuffer implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit timeUnit) { long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { while (peekNullable() == null && System.currentTimeMillis() < deadline) { lock.lock(); try { long waitTime = deadline - System.currentTimeMillis(); if (peekNullable() == null && waitTime > 0) { boolean await = dataAvailable.await(Math.min(waitTime, MAX_AWAIT_AVAILABLE_DATA), TimeUnit.MILLISECONDS); logger.trace(await ? "Signaled new events are available" : "No signal received for new events, exiting await"); } } finally { lock.unlock(); } } return peekNullable() != null; } catch (InterruptedException e) { logger.warn("Event consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } EventBuffer(EventStream delegate,
                       EventUpcaster upcasterChain,
                       Serializer serializer,
                       boolean disableEventBlacklisting); }
EventBuffer implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit timeUnit) { long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { while (peekNullable() == null && System.currentTimeMillis() < deadline) { lock.lock(); try { long waitTime = deadline - System.currentTimeMillis(); if (peekNullable() == null && waitTime > 0) { boolean await = dataAvailable.await(Math.min(waitTime, MAX_AWAIT_AVAILABLE_DATA), TimeUnit.MILLISECONDS); logger.trace(await ? "Signaled new events are available" : "No signal received for new events, exiting await"); } } finally { lock.unlock(); } } return peekNullable() != null; } catch (InterruptedException e) { logger.warn("Event consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } EventBuffer(EventStream delegate,
                       EventUpcaster upcasterChain,
                       Serializer serializer,
                       boolean disableEventBlacklisting); @Override void blacklist(TrackedEventMessage<?> trackedEventMessage); @Override Optional<TrackedEventMessage<?>> peek(); @Override boolean hasNextAvailable(int timeout, TimeUnit timeUnit); @Override TrackedEventMessage<?> nextAvailable(); @Override void close(); }
EventBuffer implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit timeUnit) { long deadline = System.currentTimeMillis() + timeUnit.toMillis(timeout); try { while (peekNullable() == null && System.currentTimeMillis() < deadline) { lock.lock(); try { long waitTime = deadline - System.currentTimeMillis(); if (peekNullable() == null && waitTime > 0) { boolean await = dataAvailable.await(Math.min(waitTime, MAX_AWAIT_AVAILABLE_DATA), TimeUnit.MILLISECONDS); logger.trace(await ? "Signaled new events are available" : "No signal received for new events, exiting await"); } } finally { lock.unlock(); } } return peekNullable() != null; } catch (InterruptedException e) { logger.warn("Event consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } EventBuffer(EventStream delegate,
                       EventUpcaster upcasterChain,
                       Serializer serializer,
                       boolean disableEventBlacklisting); @Override void blacklist(TrackedEventMessage<?> trackedEventMessage); @Override Optional<TrackedEventMessage<?>> peek(); @Override boolean hasNextAvailable(int timeout, TimeUnit timeUnit); @Override TrackedEventMessage<?> nextAvailable(); @Override void close();  }