package com.github.davidmoten.rx;

import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;

public final class SingleSubscribeSubject<T> extends Observable<T> implements Observer<T> {

    // Visible for testing
    static final String ONLY_ONE_SUBSCRIPTION_IS_ALLOWED = "only one subscription is allowed";

    private final SingleSubscribeOnSubscribe<T> subscriberHolder;

    private SingleSubscribeSubject(SingleSubscribeOnSubscribe<T> onSubscribe) {
        super(onSubscribe);
        subscriberHolder = onSubscribe;
    }

    private SingleSubscribeSubject() {
        this(new SingleSubscribeOnSubscribe<T>());
    }

    public static <T> SingleSubscribeSubject<T> create() {
        return new SingleSubscribeSubject<T>();
    }

    @Override
    public void onCompleted() {
        if (subscriberHolder.subscriber != null) {
            subscriberHolder.subscriber.onCompleted();
        }
    }

    @Override
    public void onError(Throwable e) {
        if (subscriberHolder.subscriber != null)
            subscriberHolder.subscriber.onError(e);
    }

    @Override
    public void onNext(T t) {
        if (subscriberHolder.subscriber != null)
            subscriberHolder.subscriber.onNext(t);
    }

    private static class SingleSubscribeOnSubscribe<T> implements OnSubscribe<T> {

        volatile Subscriber<? super T> subscriber;

        @SuppressWarnings("rawtypes")
        private AtomicReferenceFieldUpdater<SingleSubscribeOnSubscribe, Subscriber> SUBSCRIBER = AtomicReferenceFieldUpdater
                .newUpdater(SingleSubscribeOnSubscribe.class, Subscriber.class, "subscriber");

        @Override
        public void call(Subscriber<? super T> subscriber) {
            if (SUBSCRIBER.compareAndSet(this, null, subscriber))
                this.subscriber = subscriber;
            else
                throw new RuntimeException(ONLY_ONE_SUBSCRIPTION_IS_ALLOWED);
        }

    }

}
@@ -157,16 +157,344 @@
 criber;%0A
+import rx.subjects.Subject;%0A%0A/**%0A * A %7B@link Subject%7D that supports a maximum of one %7B@link Subscriber%7D. When%0A * there is no subscriber any notifications (%3Ccode%3EonNext%3C/code%3E,%0A * %3Ccode%3EonError%3C/code%3E, %3Ccode%3EonCompleted%3C/code%3E) are ignored.%0A * %0A * @param %3CT%3E%0A *            type of items being emitted/observed by this subject%0A */
 %0Apublic 
@@ -1017,24 +1017,147 @@
 ());%0A    %7D%0A%0A
+    /**%0A     * Returns a new instance of a %7B@link SingleSubscribeSubject%7D.%0A     * %0A     * @return the new instance%0A     */%0A
     public s

add javadoc for SingleSubscribeSubject