@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); }
@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);  }