@Test
public void testObserveOn() {

    ConnectableObservable<Integer> co = Observable.range(0, 1000).publish();
    Observable<Integer> obs = co.observeOn(Schedulers.computation());
    for (int i = 0; i < 1000; i++) {
        for (int j = 1; j < 6; j++) {
            List<TestObserver<Integer>> tos = new ArrayList<TestObserver<Integer>>();
            for (int k = 1; k < j; k++) {
                TestObserver<Integer> to = new TestObserver<Integer>();
                tos.add(to);
                obs.subscribe(to);
            }
            Disposable s = co.connect();
            for (TestObserver<Integer> to : tos) {
                to.awaitTerminalEvent(2, TimeUnit.SECONDS);
                to.assertTerminated();
                to.assertNoErrors();
                assertEquals(1000, to.valueCount());
            }
            s.dispose();
        }
    }
}