@Test public void testNoMissingBackpressureException() { final int m = 4 * 1000 * 1000; Observable<Integer> firehose = Observable.unsafeCreate(new ObservableSource<Integer>() { @Override public void subscribe(Observer<? super Integer> t) { t.onSubscribe(Disposables.empty()); for (int i = 0; i < m; i++) { t.onNext(i); } t.onComplete(); } }); TestObserver<Integer> to = new TestObserver<Integer>(); firehose.replay().autoConnect().observeOn(Schedulers.computation()).takeLast(100).subscribe(to); to.awaitTerminalEvent(3, TimeUnit.SECONDS); to.assertNoErrors(); to.assertTerminated(); assertEquals(100, to.values().size()); }
public static <T> ConnectableObservable<T> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler) { final Observable<T> observable = co.observeOn(scheduler); return RxJavaPlugins.onAssembly(new Replay<T>(co, observable)); }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { public static <T> ConnectableObservable<T> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler) { final Observable<T> observable = co.observeOn(scheduler); return RxJavaPlugins.onAssembly(new Replay<T>(co, observable)); } }
ObservableReplay extends ConnectableObservable<T> implements HasUpstreamObservableSource<T>, Disposable { public static <T> ConnectableObservable<T> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler) { final Observable<T> observable = co.observeOn(scheduler); return RxJavaPlugins.onAssembly(new Replay<T>(co, observable)); } 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> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler) { final Observable<T> observable = co.observeOn(scheduler); return RxJavaPlugins.onAssembly(new Replay<T>(co, observable)); } 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> observeOn(final ConnectableObservable<T> co, final Scheduler scheduler) { final Observable<T> observable = co.observeOn(scheduler); return RxJavaPlugins.onAssembly(new Replay<T>(co, observable)); } 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);  }