@Test
public void testTwoSubscribersOneUnsubscribes() throws Exception {

    HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-O");
    stream = HealthCountsStream.getInstance(key, 10, 100);
    final CountDownLatch latch1 = new CountDownLatch(1);
    final CountDownLatch latch2 = new CountDownLatch(1);
    final AtomicInteger healthCounts1 = new AtomicInteger(0);
    final AtomicInteger healthCounts2 = new AtomicInteger(0);
    Subscription s1 = stream
            .observe()
            .take(10)
            .observeOn(Schedulers.computation())
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    latch1.countDown();
                }
            })
            .subscribe(new Subscriber<HystrixCommandMetrics.HealthCounts>() {
                @Override
                public void onCompleted() {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 1 OnCompleted");
                    latch1.countDown();
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 1 OnError : " + e);
                    latch1.countDown();
                }
                @Override
                public void onNext(HystrixCommandMetrics.HealthCounts healthCounts) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 1 OnNext : " + healthCounts);
                    healthCounts1.incrementAndGet();
                }
            });
    Subscription s2 = stream
            .observe()
            .take(10)
            .observeOn(Schedulers.computation())
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    latch2.countDown();
                }
            })
            .subscribe(new Subscriber<HystrixCommandMetrics.HealthCounts>() {
                @Override
                public void onCompleted() {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 2 OnCompleted");
                    latch2.countDown();
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 2 OnError : " + e);
                    latch2.countDown();
                }
                @Override
                public void onNext(HystrixCommandMetrics.HealthCounts healthCounts) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 2 OnNext : " + healthCounts + " : " + healthCounts2.get());
                    healthCounts2.incrementAndGet();
                }
            });
    //execute 5 commands, then unsubscribe from first stream. then execute the rest
    for (int i = 0; i < 10; i++) {
        HystrixCommand<Integer> cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 20);
        cmd.execute();
        if (i == 5) {
            s1.unsubscribe();
        }
    }
    assertTrue(stream.isSourceCurrentlySubscribed());  //only 1/2 subscriptions has been cancelled
    assertTrue(latch1.await(10000, TimeUnit.MILLISECONDS));
    assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS));
    System.out.println("s1 got : " + healthCounts1.get() + ", s2 got : " + healthCounts2.get());
    assertTrue("s1 got data", healthCounts1.get() > 0);
    assertTrue("s2 got data", healthCounts2.get() > 0);
    assertTrue("s1 got less data than s2", healthCounts2.get() > healthCounts1.get());
}