@Test
public void testBatches() {

    HystrixCollapserKey key = HystrixCollapserKey.Factory.asKey("Collapser-Batch-Size-B");
    stream = RollingCollapserBatchSizeDistributionStream.getInstance(key, 10, 100);
    stream.startCachingStreamValuesIfUnstarted();
    final CountDownLatch latch = new CountDownLatch(1);
    stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() {
        @Override
        public void onCompleted() {
            latch.countDown();
        }
        @Override
        public void onError(Throwable e) {
            fail(e.getMessage());
        }
        @Override
        public void onNext(CachedValuesHistogram distribution) {
            System.out.println("OnNext @ " + System.currentTimeMillis());
        }
    });
    Collapser.from(key, 1).observe();
    Collapser.from(key, 2).observe();
    Collapser.from(key, 3).observe();
    try {
        Thread.sleep(250);
    } catch (InterruptedException ex) {
        fail("Interrupted ex");
    }
    Collapser.from(key, 4).observe();
    try {
        Thread.sleep(250);
    } catch (InterruptedException ex) {
        fail("Interrupted ex");
    }
    Collapser.from(key, 5).observe();
    Collapser.from(key, 6).observe();
    Collapser.from(key, 7).observe();
    Collapser.from(key, 8).observe();
    Collapser.from(key, 9).observe();
    try {
        Thread.sleep(250);
    } catch (InterruptedException ex) {
        fail("Interrupted ex");
    }
    Collapser.from(key, 10).observe();
    Collapser.from(key, 11).observe();
    Collapser.from(key, 12).observe();
    try {
        assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
    } catch (InterruptedException ex) {
        fail("Interrupted ex");
    }
    //should have 4 batches: 3, 1, 5, 3
    assertEquals(4, stream.getLatest().getTotalCount());
    assertEquals(3, stream.getLatestMean());
    assertEquals(1, stream.getLatestPercentile(0));
    assertEquals(5, stream.getLatestPercentile(100));
}