@Test public void connectRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final ConnectableObservable<Integer> co = Observable.range(1, 3).replay(); Runnable r = new Runnable() { @Override public void run() { co.connect(); } }; TestHelper.race(r, r); } }
@Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplayObserver<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf = bufferFactory.call(); ReplayObserver<T> u = new ReplayObserver<T>(buf); 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) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplayObserver<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf = bufferFactory.call(); ReplayObserver<T> u = new ReplayObserver<T>(buf); 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) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplayObserver<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf = bufferFactory.call(); ReplayObserver<T> u = new ReplayObserver<T>(buf); 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) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  ObservableReplay(ObservableSource<T> onSubscribe, ObservableSource<T> source,
                             final AtomicReference<ReplayObserver<T>> current,
                             final BufferSupplier<T> bufferFactory); }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplayObserver<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf = bufferFactory.call(); ReplayObserver<T> u = new ReplayObserver<T>(buf); 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) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  ObservableReplay(ObservableSource<T> onSubscribe, ObservableSource<T> source,
                             final AtomicReference<ReplayObserver<T>> current,
                             final BufferSupplier<T> bufferFactory); static Observable<R> multicastSelector(
            final Callable<? extends ConnectableObservable<U>> connectableFactory,
            final Function<? super Observable<U>, ? extends ObservableSource<R>> selector); static ConnectableObservable<T> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler); @SuppressWarnings("unchecked") static ConnectableObservable<T> createFrom(ObservableSource<? extends T> source); static ConnectableObservable<T> create(ObservableSource<T> source,
            final int bufferSize); static ConnectableObservable<T> create(ObservableSource<T> source,
            long maxAge, TimeUnit unit, Scheduler scheduler); static ConnectableObservable<T> create(ObservableSource<T> source,
            final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize); @Override ObservableSource<T> source(); @Override void dispose(); @Override boolean isDisposed(); @Override void connect(Consumer<? super Disposable> connection); }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { @Override public void connect(Consumer<? super Disposable> connection) { boolean doConnect; ReplayObserver<T> ps; for (;;) { ps = current.get(); if (ps == null || ps.isDisposed()) { ReplayBuffer<T> buf = bufferFactory.call(); ReplayObserver<T> u = new ReplayObserver<T>(buf); 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) { if (doConnect) { ps.shouldConnect.compareAndSet(true, false); } Exceptions.throwIfFatal(ex); throw ExceptionHelper.wrapOrThrow(ex); } if (doConnect) { source.subscribe(ps); } } private  ObservableReplay(ObservableSource<T> onSubscribe, ObservableSource<T> source,
                             final AtomicReference<ReplayObserver<T>> current,
                             final BufferSupplier<T> bufferFactory); static Observable<R> multicastSelector(
            final Callable<? extends ConnectableObservable<U>> connectableFactory,
            final Function<? super Observable<U>, ? extends ObservableSource<R>> selector); static ConnectableObservable<T> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler); @SuppressWarnings("unchecked") static ConnectableObservable<T> createFrom(ObservableSource<? extends T> source); static ConnectableObservable<T> create(ObservableSource<T> source,
            final int bufferSize); static ConnectableObservable<T> create(ObservableSource<T> source,
            long maxAge, TimeUnit unit, Scheduler scheduler); static ConnectableObservable<T> create(ObservableSource<T> source,
            final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize); @Override ObservableSource<T> source(); @Override void dispose(); @Override boolean isDisposed(); @Override void connect(Consumer<? super Disposable> connection);  }