FlowableReplay extends ConnectableFlowable<T> implements HasUpstreamPublisher<T>, Disposable { public static <T> ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize) { if (bufferSize == Integer.MAX_VALUE) { return createFrom(source); } return create(source, new ReplayBufferTask<T>(bufferSize)); } 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);  }
@Test public void testReplaySelector() { final Function<Integer, Integer> dbl = new Function<Integer, Integer>() { @Override public Integer apply(Integer t1) { return t1 * 2; } }; Function<Flowable<Integer>, Flowable<Integer>> selector = new Function<Flowable<Integer>, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Flowable<Integer> t1) { return t1.map(dbl); } }; PublishProcessor<Integer> source = PublishProcessor.create(); Flowable<Integer> co = source.replay(selector); { Subscriber<Object> observer1 = TestHelper.mockSubscriber(); InOrder inOrder = inOrder(observer1); co.subscribe(observer1); source.onNext(1); source.onNext(2); source.onNext(3); inOrder.verify(observer1, times(1)).onNext(2); inOrder.verify(observer1, times(1)).onNext(4); inOrder.verify(observer1, times(1)).onNext(6); source.onNext(4); source.onComplete(); inOrder.verify(observer1, times(1)).onNext(8); inOrder.verify(observer1, times(1)).onComplete(); inOrder.verifyNoMoreInteractions(); verify(observer1, never()).onError(any(Throwable.class)); } { Subscriber<Object> observer1 = TestHelper.mockSubscriber(); InOrder inOrder = inOrder(observer1); co.subscribe(observer1); inOrder.verify(observer1, times(1)).onComplete(); inOrder.verifyNoMoreInteractions(); verify(observer1, never()).onError(any(Throwable.class)); } }
@Test public void testBufferedReplaySelector() { final Function<Integer, Integer> dbl = new Function<Integer, Integer>() { @Override public Integer apply(Integer t1) { return t1 * 2; } }; Function<Flowable<Integer>, Flowable<Integer>> selector = new Function<Flowable<Integer>, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Flowable<Integer> t1) { return t1.map(dbl); } }; PublishProcessor<Integer> source = PublishProcessor.create(); Flowable<Integer> co = source.replay(selector, 3); { Subscriber<Object> observer1 = TestHelper.mockSubscriber(); InOrder inOrder = inOrder(observer1); co.subscribe(observer1); source.onNext(1); source.onNext(2); source.onNext(3); inOrder.verify(observer1, times(1)).onNext(2); inOrder.verify(observer1, times(1)).onNext(4); inOrder.verify(observer1, times(1)).onNext(6); source.onNext(4); source.onComplete(); inOrder.verify(observer1, times(1)).onNext(8); inOrder.verify(observer1, times(1)).onComplete(); inOrder.verifyNoMoreInteractions(); verify(observer1, never()).onError(any(Throwable.class)); } { Subscriber<Object> observer1 = TestHelper.mockSubscriber(); InOrder inOrder = inOrder(observer1); co.subscribe(observer1); inOrder.verify(observer1, times(1)).onComplete(); inOrder.verifyNoMoreInteractions(); verify(observer1, never()).onError(any(Throwable.class)); } }
@Test public void testWindowedReplaySelector() { final Function<Integer, Integer> dbl = new Function<Integer, Integer>() { @Override public Integer apply(Integer t1) { return t1 * 2; } }; Function<Flowable<Integer>, Flowable<Integer>> selector = new Function<Flowable<Integer>, Flowable<Integer>>() { @Override public Flowable<Integer> apply(Flowable<Integer> t1) { return t1.map(dbl); } }; TestScheduler scheduler = new TestScheduler(); PublishProcessor<Integer> source = PublishProcessor.create(); Flowable<Integer> co = source.replay(selector, 100, TimeUnit.MILLISECONDS, scheduler); { Subscriber<Object> observer1 = TestHelper.mockSubscriber(); InOrder inOrder = inOrder(observer1); co.subscribe(observer1); source.onNext(1); scheduler.advanceTimeBy(60, TimeUnit.MILLISECONDS); source.onNext(2); scheduler.advanceTimeBy(60, TimeUnit.MILLISECONDS); source.onNext(3); scheduler.advanceTimeBy(60, TimeUnit.MILLISECONDS); source.onComplete(); scheduler.advanceTimeBy(60, TimeUnit.MILLISECONDS); inOrder.verify(observer1, times(1)).onNext(2); inOrder.verify(observer1, times(1)).onNext(4); inOrder.verify(observer1, times(1)).onNext(6); inOrder.verify(observer1, times(1)).onComplete(); inOrder.verifyNoMoreInteractions(); verify(observer1, never()).onError(any(Throwable.class)); } { Subscriber<Object> observer1 = TestHelper.mockSubscriber(); InOrder inOrder = inOrder(observer1); co.subscribe(observer1); inOrder.verify(observer1, times(1)).onComplete(); inOrder.verifyNoMoreInteractions(); verify(observer1, never()).onError(any(Throwable.class)); } }
@Test public void cancelOnArrival2() { ConnectableFlowable<Integer> cf = PublishProcessor.<Integer>create() .replay(Integer.MAX_VALUE); cf.test(); cf .autoConnect() .test(Long.MAX_VALUE, true) .assertEmpty(); }
@Test public void subscribeOnNextRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { final PublishProcessor<Integer> pp = PublishProcessor.create(); final ConnectableFlowable<Integer> cf = pp.replay(); final TestSubscriber<Integer> ts1 = new TestSubscriber<Integer>(); Runnable r1 = new Runnable() { @Override public void run() { cf.subscribe(ts1); } }; Runnable r2 = new Runnable() { @Override public void run() { for (int j = 0; j < 1000; j++) { pp.onNext(j); } } }; TestHelper.race(r1, r2); } }
@Test public void reentrantOnNext() { final PublishProcessor<Integer> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { if (t == 1) { pp.onNext(2); pp.onComplete(); } super.onNext(t); } }; pp.replay().autoConnect().subscribe(ts); pp.onNext(1); ts.assertResult(1, 2); }
@Test public void reentrantOnNextBound() { final PublishProcessor<Integer> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { if (t == 1) { pp.onNext(2); pp.onComplete(); } super.onNext(t); } }; pp.replay(10).autoConnect().subscribe(ts); pp.onNext(1); ts.assertResult(1, 2); }
@Test public void reentrantOnNextCancel() { final PublishProcessor<Integer> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { if (t == 1) { pp.onNext(2); cancel(); } super.onNext(t); } }; pp.replay().autoConnect().subscribe(ts); pp.onNext(1); ts.assertValues(1); }
@Test public void reentrantOnNextCancelBounded() { final PublishProcessor<Integer> pp = PublishProcessor.create(); TestSubscriber<Integer> ts = new TestSubscriber<Integer>() { @Override public void onNext(Integer t) { if (t == 1) { pp.onNext(2); cancel(); } super.onNext(t); } }; pp.replay(10).autoConnect().subscribe(ts); pp.onNext(1); ts.assertValues(1); }
@Test public void createBufferFactoryCrashOnSubscribe() { FlowableReplay.create(Flowable.just(1), new Callable<ReplayBuffer<Integer>>() { @Override public ReplayBuffer<Integer> call() throws Exception { throw new TestException(); } }) .test() .assertFailure(TestException.class); }