ThresholdBatcher { public void add(E e) throws FlowControlException { flowController.reserve(e); lock.lock(); try { receiver.validateBatch(e); boolean anyThresholdReached = isAnyThresholdReached(e); if (currentOpenBatch == null) { currentOpenBatch = e; if (!anyThresholdReached) { currentAlarmFuture = executor.schedule( pushCurrentBatchRunnable, maxDelay.toMillis(), TimeUnit.MILLISECONDS); } } else { batchMerger.merge(currentOpenBatch, e); } if (anyThresholdReached) { pushCurrentBatch(); } } finally { lock.unlock(); } } private  ThresholdBatcher(Builder<E> builder); static Builder<E> newBuilder(); void add(E e); @VisibleForTesting ApiFuture<Void> pushCurrentBatch();  }
@Test public void testAdd() throws Exception { AccumulatingBatchReceiver<SimpleBatch> receiver = new AccumulatingBatchReceiver<>(ApiFutures.<Void>immediateFuture(null)); ThresholdBatcher<SimpleBatch> batcher = createSimpleBatcherBuidler(receiver).build(); batcher.add(SimpleBatch.fromInteger(14)); assertThat(batcher.isEmpty()).isFalse(); assertThat(receiver.getBatches()).hasSize(0); batcher.pushCurrentBatch().get(); assertThat(batcher.isEmpty()).isTrue(); assertThat(receiver.getBatches()).hasSize(1); assertThat(receiver.getBatches().get(0).getIntegers()).isEqualTo(Arrays.asList(14)); }