@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));
    }
}