Operator Class

- forgot to add earlier

Author: benjchristensen

rxjava-core/src/main/java/rx/Operator.java

@@ -0,0 +1,70 @@
+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();
+    }
+}