class OperatorPublish extends ConnectableObservable<T> {

    @Override
    public void connect(Action1<? super Subscription> connection) {
        // each time we connect we create a new Subscription
        boolean shouldSubscribe = false;
        
        // subscription is the state of whether we are connected or not
        OriginSubscriber<T> origin = requestHandler.state.getOrigin();
        if (origin == null) {
            shouldSubscribe = true;
            requestHandler.state.setOrigin(new OriginSubscriber<T>(requestHandler));
        }

        // in the lock above we determined we should subscribe, do it now outside the lock
        if (shouldSubscribe) {
            // register a subscription that will shut this down
            connection.call(Subscriptions.create(new Action0() {
                @Override
                public void call() {
                    OriginSubscriber<T> s = requestHandler.state.getOrigin();
                    requestHandler.state.setOrigin(null);
                    if (s != null) {
                        s.unsubscribe();
                    }
                }
            }));

            // now that everything is hooked up let's subscribe
            // as long as the subscription is not null (which can happen if already unsubscribed)
            OriginSubscriber<T> os = requestHandler.state.getOrigin();
            if (os != null) {
                source.unsafeSubscribe(os);
            }
        }
    }

    private  OperatorPublish(Observable<? extends T> source);
    private  OperatorPublish(Observable<? extends T> source, final Object guard, final RequestHandler<T> requestHandler);

}