@Test public void subscribeOnNextRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final PublishSubject<Integer> ps = PublishSubject.create(); final ConnectableObservable<Integer> co = ps.replay(); final TestObserver<Integer> to1 = new TestObserver<Integer>(); Runnable r1 = new Runnable() { @Override public void run() { co.subscribe(to1); } }; Runnable r2 = new Runnable() { @Override public void run() { for (int j = 0; j < 1000; j++) { ps.onNext(j); } } }; TestHelper.race(r1, r2); } }
public static <T> ConnectableObservable<T> create(ObservableSource<T> source, final int bufferSize) { if (bufferSize == Integer.MAX_VALUE) { return createFrom(source); } return create(source, new ReplayBufferSupplier<T>(bufferSize)); }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { public static <T> ConnectableObservable<T> create(ObservableSource<T> source, final int bufferSize) { if (bufferSize == Integer.MAX_VALUE) { return createFrom(source); } return create(source, new ReplayBufferSupplier<T>(bufferSize)); } }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { public static <T> ConnectableObservable<T> create(ObservableSource<T> source, final int bufferSize) { if (bufferSize == Integer.MAX_VALUE) { return createFrom(source); } return create(source, new ReplayBufferSupplier<T>(bufferSize)); } 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 { public static <T> ConnectableObservable<T> create(ObservableSource<T> source, final int bufferSize) { if (bufferSize == Integer.MAX_VALUE) { return createFrom(source); } return create(source, new ReplayBufferSupplier<T>(bufferSize)); } 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 { public static <T> ConnectableObservable<T> create(ObservableSource<T> source, final int bufferSize) { if (bufferSize == Integer.MAX_VALUE) { return createFrom(source); } return create(source, new ReplayBufferSupplier<T>(bufferSize)); } 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);  }