@Test
public void testSharedSourceStream() throws InterruptedException {

    HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-N");
    stream = HealthCountsStream.getInstance(key, 10, 100);
    final CountDownLatch latch = new CountDownLatch(1);
    final AtomicBoolean allEqual = new AtomicBoolean(false);
    Observable<HystrixCommandMetrics.HealthCounts> o1 = stream
            .observe()
            .take(10)
            .observeOn(Schedulers.computation());
    Observable<HystrixCommandMetrics.HealthCounts> o2 = stream
            .observe()
            .take(10)
            .observeOn(Schedulers.computation());
    Observable<Boolean> zipped = Observable.zip(o1, o2, new Func2<HystrixCommandMetrics.HealthCounts, HystrixCommandMetrics.HealthCounts, Boolean>() {
        @Override
        public Boolean call(HystrixCommandMetrics.HealthCounts healthCounts, HystrixCommandMetrics.HealthCounts healthCounts2) {
            return healthCounts == healthCounts2;  //we want object equality
        }
    });
    Observable < Boolean > reduced = zipped.reduce(true, new Func2<Boolean, Boolean, Boolean>() {
        @Override
        public Boolean call(Boolean a, Boolean b) {
            return a && b;
        }
    });
    reduced.subscribe(new Subscriber<Boolean>() {
        @Override
        public void onCompleted() {
            System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Reduced OnCompleted");
            latch.countDown();
        }
        @Override
        public void onError(Throwable e) {
            System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Reduced OnError : " + e);
            e.printStackTrace();
            latch.countDown();
        }
        @Override
        public void onNext(Boolean b) {
            System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Reduced OnNext : " + b);
            allEqual.set(b);
        }
    });
    for (int i = 0; i < 10; i++) {
        HystrixCommand<Integer> cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 20);
        cmd.execute();
    }
    assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
    assertTrue(allEqual.get());
    //we should be getting the same object from both streams.  this ensures that multiple subscribers don't induce extra work
}