@Test
public void testTwoSubscribersOneUnsubscribes() throws Exception {

    final CountDownLatch latch1 = new CountDownLatch(1);
    final CountDownLatch latch2 = new CountDownLatch(1);
    final AtomicInteger payloads1 = new AtomicInteger(0);
    final AtomicInteger payloads2 = new AtomicInteger(0);
    Subscription s1 = stream
            .observe()
            .take(100)
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    latch1.countDown();
                }
            })
            .subscribe(new Subscriber<HystrixUtilization>() {
                @Override
                public void onCompleted() {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Dashboard 1 OnCompleted");
                    latch1.countDown();
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Dashboard 1 OnError : " + e);
                    latch1.countDown();
                }
                @Override
                public void onNext(HystrixUtilization utilization) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Dashboard 1 OnNext : " + utilization);
                    payloads1.incrementAndGet();
                }
            });
    Subscription s2 = stream
            .observe()
            .take(100)
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    latch2.countDown();
                }
            })
            .subscribe(new Subscriber<HystrixUtilization>() {
                @Override
                public void onCompleted() {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Dashboard 2 OnCompleted");
                    latch2.countDown();
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Dashboard 2 OnError : " + e);
                    latch2.countDown();
                }
                @Override
                public void onNext(HystrixUtilization utilization) {
                    System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Dashboard 2 OnNext : " + utilization);
                    payloads2.incrementAndGet();
                }
            });
    //execute 1 command, then unsubscribe from first stream. then execute the rest
    for (int i = 0; i < 50; i++) {
        HystrixCommand<Integer> cmd = Command.from(groupKey, commandKey, HystrixEventType.SUCCESS, 50);
        cmd.execute();
        if (i == 1) {
            s1.unsubscribe();
        }
    }
    assertTrue(latch1.await(10000, TimeUnit.MILLISECONDS));
    assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS));
    System.out.println("s1 got : " + payloads1.get() + ", s2 got : " + payloads2.get());
    assertTrue("s1 got data", payloads1.get() > 0);
    assertTrue("s2 got data", payloads2.get() > 0);
    assertTrue("s1 got less data than s2", payloads2.get() > payloads1.get());
}