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

}

class FlowableReplayTest {

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

}