@Test public void stormTest() { TestAccumulator accumulator = new TestAccumulator(); IntStream.range(0, 1000).forEach(i -> accumulator.add(new TestItem("#" + i))); timer.advanceTimeMillis(1); assertAfter(100, () -> assertEquals("wrong item count", 1000, accumulator.itemCount)); assertEquals("wrong batch count", 200, accumulator.batchCount); }
@Override public synchronized void add(T item) { idleTask = cancelIfActive(idleTask); items.add(checkNotNull(item, "Item cannot be null")); if (items.size() >= maxItems) { maxTask = cancelIfActive(maxTask); scheduleNow(); } else { idleTask = schedule(maxIdleMillis); if (items.size() == 1) { maxTask = schedule(maxBatchMillis); } } }
AbstractAccumulator implements Accumulator<T> { @Override public synchronized void add(T item) { idleTask = cancelIfActive(idleTask); items.add(checkNotNull(item, "Item cannot be null")); if (items.size() >= maxItems) { maxTask = cancelIfActive(maxTask); scheduleNow(); } else { idleTask = schedule(maxIdleMillis); if (items.size() == 1) { maxTask = schedule(maxBatchMillis); } } } }
AbstractAccumulator implements Accumulator<T> { @Override public synchronized void add(T item) { idleTask = cancelIfActive(idleTask); items.add(checkNotNull(item, "Item cannot be null")); if (items.size() >= maxItems) { maxTask = cancelIfActive(maxTask); scheduleNow(); } else { idleTask = schedule(maxIdleMillis); if (items.size() == 1) { maxTask = schedule(maxBatchMillis); } } } protected  AbstractAccumulator(Timer timer, int maxItems,
                                  int maxBatchMillis, int maxIdleMillis); }
AbstractAccumulator implements Accumulator<T> { @Override public synchronized void add(T item) { idleTask = cancelIfActive(idleTask); items.add(checkNotNull(item, "Item cannot be null")); if (items.size() >= maxItems) { maxTask = cancelIfActive(maxTask); scheduleNow(); } else { idleTask = schedule(maxIdleMillis); if (items.size() == 1) { maxTask = schedule(maxBatchMillis); } } } protected  AbstractAccumulator(Timer timer, int maxItems,
                                  int maxBatchMillis, int maxIdleMillis); @Override synchronized void add(T item); @Override boolean isReady(); Timer timer(); int maxItems(); int maxBatchMillis(); int maxIdleMillis(); }
AbstractAccumulator implements Accumulator<T> { @Override public synchronized void add(T item) { idleTask = cancelIfActive(idleTask); items.add(checkNotNull(item, "Item cannot be null")); if (items.size() >= maxItems) { maxTask = cancelIfActive(maxTask); scheduleNow(); } else { idleTask = schedule(maxIdleMillis); if (items.size() == 1) { maxTask = schedule(maxBatchMillis); } } } protected  AbstractAccumulator(Timer timer, int maxItems,
                                  int maxBatchMillis, int maxIdleMillis); @Override synchronized void add(T item); @Override boolean isReady(); Timer timer(); int maxItems(); int maxBatchMillis(); int maxIdleMillis();  }