CapacityBasedBufferMonitor implements IBufferMonitor<T> { @Override public void eventAdded(final T event, final IFlushAndPublish cache) { boolean flush = false; synchronized (countGuard) { if (++count >= cacheLimit) { flush = true; count = 0; } } if (flush) { try { cache.flushAndPublish(); } catch (Exception ex) { ex.printStackTrace(); } } } CapacityBasedBufferMonitor(int cacheLimit); @Override void eventAdded(final T event, final IFlushAndPublish cache); @Override void shutDown(); @Override String toString();  }
@Test @SuppressWarnings("unchecked") public void testFlush() { CapacityBasedBufferMonitor monitor = new CapacityBasedBufferMonitor(5); for (int i = 0; i < 25; i++) { monitor.eventAdded(null, publisher); } Assert.assertEquals("5 batches of 5 events per batch", 5, publishCount); }