@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); } }
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)); }
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)); } }
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); }
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); }
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);  }