@Test public void testObserveOn() { ConnectableFlowable<Integer> cf = Flowable.range(0, 1000).hide().publish(); Flowable<Integer> obs = cf.observeOn(Schedulers.computation()); for (int i = 0; i < 1000; i++) { for (int j = 1; j < 6; j++) { List<TestSubscriber<Integer>> tss = new ArrayList<TestSubscriber<Integer>>(); for (int k = 1; k < j; k++) { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); tss.add(ts); obs.subscribe(ts); } Disposable s = cf.connect(); for (TestSubscriber<Integer> ts : tss) { ts.awaitDone(5, TimeUnit.SECONDS) .assertSubscribed() .assertValueCount(1000) .assertNoErrors() .assertComplete(); } s.dispose(); } } }
@Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; PublishSubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { PublishSubscriber<T> u = new PublishSubscriber<T>(current, bufferSize); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } }
FlowablePublish extends ConnectableFlowable<T> implements HasUpstreamPublisher<T> { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; PublishSubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { PublishSubscriber<T> u = new PublishSubscriber<T>(current, bufferSize); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } }
FlowablePublish extends ConnectableFlowable<T> implements HasUpstreamPublisher<T> { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; PublishSubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { PublishSubscriber<T> u = new PublishSubscriber<T>(current, bufferSize); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  FlowablePublish(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<PublishSubscriber<T>> current, int bufferSize); }
FlowablePublish extends ConnectableFlowable<T> implements HasUpstreamPublisher<T> { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; PublishSubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { PublishSubscriber<T> u = new PublishSubscriber<T>(current, bufferSize); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  FlowablePublish(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<PublishSubscriber<T>> current, int bufferSize); static ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize); @Override Publisher<T> source(); @Override void connect(Consumer<? super Disposable> connection); }
FlowablePublish extends ConnectableFlowable<T> implements HasUpstreamPublisher<T> { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; PublishSubscriber<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { PublishSubscriber<T> u = new PublishSubscriber<T>(current, bufferSize); if (!current.compareAndSet(ps, u)) { continue; } ps = u; } doConnect = !ps.shouldConnect.get() && ps.shouldConnect.compareAndSet(false, true); break; } try { connection.accept(ps); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  FlowablePublish(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<PublishSubscriber<T>> current, int bufferSize); static ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize); @Override Publisher<T> source(); @Override void connect(Consumer<? super Disposable> connection);  }