package rx;

import rx.subscriptions.CompositeSubscription;

public abstract class Operator<T> implements Observer<T>, Subscription {

    private final CompositeSubscription cs;

    // TODO I'm questioning this API, it could be confusing and misused
    protected Operator(Operator<?> op) {
        this.cs = op.cs;
    }

    protected Operator(CompositeSubscription cs) {
        this.cs = cs;
    }

    public static <T> Operator<T> create(final Observer<? super T> o, CompositeSubscription cs) {
        if (o == null) {
            throw new IllegalArgumentException("Observer can not be null");
        }
        if (cs == null) {
            throw new IllegalArgumentException("CompositeSubscription can not be null");
        }
        return new Operator<T>(cs) {

            @Override
            public void onCompleted() {
                o.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                o.onError(e);
            }

            @Override
            public void onNext(T v) {
                o.onNext(v);
            }

        };
    }

    public static <T> Operator<T> create(final Observer<? super T> o, Subscription s) {
        if (s == null) {
            throw new IllegalArgumentException("Subscription can not be null");
        }
        CompositeSubscription cs = new CompositeSubscription();
        cs.add(s);

        return create(o, cs);
    }

    /**
     * Used to register an unsubscribe callback.
     */
    public final void add(Subscription s) {
        cs.add(s);
    }

    @Override
    public final void unsubscribe() {
        cs.unsubscribe();
    }

    public final boolean isUnsubscribed() {
        return cs.isUnsubscribed();
    }
}