public static <T> ConnectableFlowable<T> create(Flowable<T> source, final int bufferSize) {
        // the current connection to source needs to be shared between the operator and its onSubscribe call
        final AtomicReference<PublishSubscriber<T>> curr = new AtomicReference<PublishSubscriber<T>>();
        Publisher<T> onSubscribe = new FlowablePublisher<T>(curr, bufferSize);
        return RxJavaPlugins.onAssembly(new FlowablePublish<T>(onSubscribe, source, curr, bufferSize));
    }