@Test public void handlesEventProcessorExceptions() throws Exception { for (int i = 0; i < 1000; i++) { CyclicBarrier barrier1 = new CyclicBarrier(2); CyclicBarrier barrier2 = new CyclicBarrier(2); CyclicBarrier barrier3 = new CyclicBarrier(2); QueueDrainingEventProcessor eventProcessor = new QueueDrainingEventProcessor((event) -> ((Consumer<Void>) event).accept(null), false); startThread( () -> { Consumer<Void> lockEvent = consumerEvent((x) -> { await(barrier1); try { throw new JustATestException(); } finally { await(barrier2); } }); eventProcessor.submit(lockEvent); }); barrier1.await(); Consumer<Void> event2 = mock(Consumer.class); eventProcessor.submit(consumerEvent(x -> { event2.accept(null); await(barrier3); })); await(barrier2); await(barrier3); verify(event2).accept(eq(null)); } }
@Override public void submit(Object event) { events.add(event); if (eventCount.getAndIncrement() == 0) { do { Object e = events.poll(); try { eventProcessor.submit(e); } catch (RuntimeException cause) { if (logErrors) { LOGGER.warn("Event {} threw an exception {}.", event, cause); } } } while (eventCount.decrementAndGet() > 0); } }
QueueDrainingEventProcessor implements EventProcessor { @Override public void submit(Object event) { events.add(event); if (eventCount.getAndIncrement() == 0) { do { Object e = events.poll(); try { eventProcessor.submit(e); } catch (RuntimeException cause) { if (logErrors) { LOGGER.warn("Event {} threw an exception {}.", event, cause); } } } while (eventCount.decrementAndGet() > 0); } } }
QueueDrainingEventProcessor implements EventProcessor { @Override public void submit(Object event) { events.add(event); if (eventCount.getAndIncrement() == 0) { do { Object e = events.poll(); try { eventProcessor.submit(e); } catch (RuntimeException cause) { if (logErrors) { LOGGER.warn("Event {} threw an exception {}.", event, cause); } } } while (eventCount.decrementAndGet() > 0); } } QueueDrainingEventProcessor(EventProcessor eventProcessor);  QueueDrainingEventProcessor(EventProcessor eventProcessor, boolean logErrors); }
QueueDrainingEventProcessor implements EventProcessor { @Override public void submit(Object event) { events.add(event); if (eventCount.getAndIncrement() == 0) { do { Object e = events.poll(); try { eventProcessor.submit(e); } catch (RuntimeException cause) { if (logErrors) { LOGGER.warn("Event {} threw an exception {}.", event, cause); } } } while (eventCount.decrementAndGet() > 0); } } QueueDrainingEventProcessor(EventProcessor eventProcessor);  QueueDrainingEventProcessor(EventProcessor eventProcessor, boolean logErrors); @Override void submit(Object event); }
QueueDrainingEventProcessor implements EventProcessor { @Override public void submit(Object event) { events.add(event); if (eventCount.getAndIncrement() == 0) { do { Object e = events.poll(); try { eventProcessor.submit(e); } catch (RuntimeException cause) { if (logErrors) { LOGGER.warn("Event {} threw an exception {}.", event, cause); } } } while (eventCount.decrementAndGet() > 0); } } QueueDrainingEventProcessor(EventProcessor eventProcessor);  QueueDrainingEventProcessor(EventProcessor eventProcessor, boolean logErrors); @Override void submit(Object event);  }