public static <T> ConnectableObservable<T> create(ObservableSource<T> source) {
        // the current connection to source needs to be shared between the operator and its onSubscribe call
        final AtomicReference<PublishObserver<T>> curr = new AtomicReference<PublishObserver<T>>();
        ObservableSource<T> onSubscribe = new PublishSource<T>(curr);
        return RxJavaPlugins.onAssembly(new ObservablePublish<T>(onSubscribe, source, curr));
    }