@Test public void testAsyncComeAndGo() { Flowable<Long> source = Flowable.interval(1, 1, TimeUnit.MILLISECONDS) .take(1000) .subscribeOn(Schedulers.io()); Flowable<Long> cached = source.replay().autoConnect(); Flowable<Long> output = cached.observeOn(Schedulers.computation(), false, 1024); List<TestSubscriber<Long>> list = new ArrayList<TestSubscriber<Long>>(100); for (int i = 0; i < 100; i++) { TestSubscriber<Long> ts = new TestSubscriber<Long>(); list.add(ts); output.skip(i * 10).take(10).subscribe(ts); } List<Long> expected = new ArrayList<Long>(); for (int i = 0; i < 10; i++) { expected.add((long)(i - 10)); } int j = 0; for (TestSubscriber<Long> ts : list) { ts.awaitTerminalEvent(3, TimeUnit.SECONDS); ts.assertNoErrors(); ts.assertTerminated(); for (int i = j * 10; i < j * 10 + 10; i++) { expected.set(i - j * 10, (long)i); } ts.assertValueSequence(expected); j++; } }
public static <T> ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler) { final Flowable<T> observable = cf.observeOn(scheduler); return RxJavaPlugins.onAssembly(new ConnectableFlowableReplay<T>(cf, observable)); }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { public static <T> ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler) { final Flowable<T> observable = cf.observeOn(scheduler); return RxJavaPlugins.onAssembly(new ConnectableFlowableReplay<T>(cf, observable)); } }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { public static <T> ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler) { final Flowable<T> observable = cf.observeOn(scheduler); return RxJavaPlugins.onAssembly(new ConnectableFlowableReplay<T>(cf, observable)); } private  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<ReplaySubscriber<T>> current,
            final Callable<? extends ReplayBuffer<T>> bufferFactory); }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { public static <T> ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler) { final Flowable<T> observable = cf.observeOn(scheduler); return RxJavaPlugins.onAssembly(new ConnectableFlowableReplay<T>(cf, observable)); } private  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<ReplaySubscriber<T>> current,
            final Callable<? extends ReplayBuffer<T>> bufferFactory); static Flowable<R> multicastSelector(
            final Callable<? extends ConnectableFlowable<U>> connectableFactory,
            final Function<? super Flowable<U>, ? extends Publisher<R>> selector); static ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler); @SuppressWarnings("unchecked") static ConnectableFlowable<T> createFrom(Flowable<? extends T> source); static ConnectableFlowable<T> create(Flowable<T> source,
            final int bufferSize); static ConnectableFlowable<T> create(Flowable<T> source,
            long maxAge, TimeUnit unit, Scheduler scheduler); static ConnectableFlowable<T> create(Flowable<T> source,
            final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize); @Override Publisher<T> source(); @Override void dispose(); @Override boolean isDisposed(); @Override void connect(Consumer<? super Disposable> connection); }
FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { public static <T> ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler) { final Flowable<T> observable = cf.observeOn(scheduler); return RxJavaPlugins.onAssembly(new ConnectableFlowableReplay<T>(cf, observable)); } private  FlowableReplay(Publisher<T> onSubscribe, Flowable<T> source,
            final AtomicReference<ReplaySubscriber<T>> current,
            final Callable<? extends ReplayBuffer<T>> bufferFactory); static Flowable<R> multicastSelector(
            final Callable<? extends ConnectableFlowable<U>> connectableFactory,
            final Function<? super Flowable<U>, ? extends Publisher<R>> selector); static ConnectableFlowable<T> observeOn(final ConnectableFlowable<T> cf, final Scheduler scheduler); @SuppressWarnings("unchecked") static ConnectableFlowable<T> createFrom(Flowable<? extends T> source); static ConnectableFlowable<T> create(Flowable<T> source,
            final int bufferSize); static ConnectableFlowable<T> create(Flowable<T> source,
            long maxAge, TimeUnit unit, Scheduler scheduler); static ConnectableFlowable<T> create(Flowable<T> source,
            final long maxAge, final TimeUnit unit, final Scheduler scheduler, final int bufferSize); @Override Publisher<T> source(); @Override void dispose(); @Override boolean isDisposed(); @Override void connect(Consumer<? super Disposable> connection);  }