RollingCollapserBatchSizeDistributionStream extends RollingDistributionStream<HystrixCollapserEvent> { public static RollingCollapserBatchSizeDistributionStream getInstance(HystrixCollapserKey collapserKey, HystrixCollapserProperties properties) { final int percentileMetricWindow = properties.metricsRollingPercentileWindowInMilliseconds().get(); final int numPercentileBuckets = properties.metricsRollingPercentileWindowBuckets().get(); final int percentileBucketSizeInMs = percentileMetricWindow / numPercentileBuckets; return getInstance(collapserKey, numPercentileBuckets, percentileBucketSizeInMs); } private  RollingCollapserBatchSizeDistributionStream(HystrixCollapserKey collapserKey, int numPercentileBuckets, int percentileBucketSizeInMs); static RollingCollapserBatchSizeDistributionStream getInstance(HystrixCollapserKey collapserKey, HystrixCollapserProperties properties); static RollingCollapserBatchSizeDistributionStream getInstance(HystrixCollapserKey collapserKey, int numBuckets, int bucketSizeInMs); static void reset();  }
@Test public void testEmptyStreamProducesEmptyDistributions() { HystrixCollapserKey key = HystrixCollapserKey.Factory.asKey("Collapser-Batch-Size-A"); stream = RollingCollapserBatchSizeDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().skip(10).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()); assertEquals(0, distribution.getTotalCount()); } }); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(0, stream.getLatest().getTotalCount()); }
@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"); } assertEquals(4, stream.getLatest().getTotalCount()); assertEquals(3, stream.getLatestMean()); assertEquals(1, stream.getLatestPercentile(0)); assertEquals(5, stream.getLatestPercentile(100)); }
@Test public void testBatchesAgeOut() { 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(30).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(200); } catch (InterruptedException ex) { fail("Interrupted ex"); } Collapser.from(key, 4).observe(); try { Thread.sleep(200); } 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(200); } 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"); } assertEquals(0, stream.getLatest().getTotalCount()); assertEquals(0, stream.getLatestMean()); }