ObservablePublish extends ConnectableObservable<T> implements HasUpstreamObservableSource<T> { public static <T> ConnectableObservable<T> create(ObservableSource<T> source) { final AtomicReference<PublishObserver<T>> curr = new AtomicReference<PublishObserver<T>>(); ObservableSource<T> onSubscribe = new PublishSource<T>(curr); return RxJavaPlugins.onAssembly(new ObservablePublish<T>(onSubscribe, source, curr)); } private  ObservablePublish(ObservableSource<T> onSubscribe, ObservableSource<T> source,
                              final AtomicReference<PublishObserver<T>> current); static ConnectableObservable<T> create(ObservableSource<T> source); @Override ObservableSource<T> source(); @Override void connect(Consumer<? super Disposable> connection);  }
@Test public void nextCancelRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final PublishSubject<Integer> ps = PublishSubject.create(); final ConnectableObservable<Integer> co = ps.publish(); final TestObserver<Integer> to = co.test(); Runnable r1 = new Runnable() { @Override public void run() { ps.onNext(1); } }; Runnable r2 = new Runnable() { @Override public void run() { to.cancel(); } }; TestHelper.race(r1, r2); } }
@Test public void selectorDisconnectsIndependentSource() { PublishSubject<Integer> ps = PublishSubject.create(); ps.publish(new Function<Observable<Integer>, ObservableSource<Integer>>() { @Override public ObservableSource<Integer> apply(Observable<Integer> v) throws Exception { return Observable.range(1, 2); } }) .test() .assertResult(1, 2); assertFalse(ps.hasObservers()); }
@Test public void selectorInnerError() { PublishSubject<Integer> ps = PublishSubject.create(); ps.publish(new Function<Observable<Integer>, ObservableSource<Integer>>() { @Override public ObservableSource<Integer> apply(Observable<Integer> v) throws Exception { return Observable.error(new TestException()); } }) .test() .assertFailure(TestException.class); assertFalse(ps.hasObservers()); }