@Test public void processesQueuedEvents() { for (int i = 0; i < 1000; i++) { CyclicBarrier barrier1 = new CyclicBarrier(2); CyclicBarrier barrier2 = new CyclicBarrier(2); QueueDrainingEventProcessor eventProcessor = new QueueDrainingEventProcessor((event) -> ((Consumer<Void>) event).accept(null)); Consumer<Void> event1 = mock(Consumer.class); Consumer<Void> event2 = mock(Consumer.class); Consumer<Void> event3 = mock(Consumer.class); startThread(() -> { await(barrier1); eventProcessor.submit(event2); eventProcessor.submit(event3); await(barrier2); }); eventProcessor.submit(consumerEvent((x) -> { await(barrier1); event1.accept(null); await(barrier2); })); InOrder inOrder = Mockito.inOrder(event1, event2, event3); inOrder.verify(event1).accept(null); inOrder.verify(event2).accept(null); inOrder.verify(event3).accept(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);  }