src/main/java/io/reactivex/internal/operators/flowable/FlowableReplay.java
static <T> ConnectableFlowable<T> create(Flowable<T> source,
            final Callable<? extends ReplayBuffer<T>> bufferFactory) {
        // the current connection to source needs to be shared between the operator and its onSubscribe call
        final AtomicReference<ReplaySubscriber<T>> curr = new AtomicReference<ReplaySubscriber<T>>();
        Publisher<T> onSubscribe = new ReplayPublisher<T>(curr, bufferFactory);
        return RxJavaPlugins.onAssembly(new FlowableReplay<T>(onSubscribe, source, curr, bufferFactory));
    }
static <T> ConnectableFlowable<T> create(Flowable<T> source,
            final Callable<? extends ReplayBuffer<T>> bufferFactory) {
        // the current connection to source needs to be shared between the operator and its onSubscribe call
        final AtomicReference<ReplaySubscriber<T>> curr = new AtomicReference<ReplaySubscriber<T>>();
        Publisher<T> onSubscribe = new ReplayPublisher<T>(curr, bufferFactory);
        return RxJavaPlugins.onAssembly(new FlowableReplay<T>(onSubscribe, source, curr, bufferFactory));
    }
static < T > ConnectableFlowable < T > create ( Flowable < T > source , final Callable < ? extends ReplayBuffer < T > > bufferFactory ) { // the current connection to source needs to be shared between the operator and its onSubscribe call final AtomicReference < ReplaySubscriber < T > > curr = new AtomicReference < ReplaySubscriber < T > > ( ) ; Publisher < T > onSubscribe = new ReplayPublisher < T > ( curr , bufferFactory ) ; return RxJavaPlugins . onAssembly ( new FlowableReplay < T > ( onSubscribe , source , curr , bufferFactory ) ) ; }
Creates a OperatorReplay instance to replay values of the given source observable.
@param source the source observable
@param bufferFactory the factory to instantiate the appropriate buffer when the observable becomes active
@return the connectable observable
Creates a OperatorReplay instance to replay values of the given source observable .