### Question:
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);  }### Answer:
@Test public void processesEvents() { QueueDrainingEventProcessor eventProcessor = new QueueDrainingEventProcessor((event) -> ((Consumer<Void>) event).accept(null)); Consumer<Void> event1 = mock(Consumer.class); eventProcessor.submit(event1); Consumer<Void> event2 = mock(Consumer.class); eventProcessor.submit(event2); InOrder inOrder = Mockito.inOrder(event1, event2); inOrder.verify(event1).accept(null); inOrder.verify(event2).accept(null); }
@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); } }
@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)); } }