class ObservablePublish extends ConnectableObservable<T> implements HasUpstreamObservableSource<T> {

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

    private  ObservablePublish(ObservableSource<T> onSubscribe, ObservableSource<T> source,
                              final AtomicReference<PublishObserver<T>> current);

}