Subscriber and Observer

-> Restore Observer interface
-> Subscriber implements Observer, Subscription

Author: benjchristensen

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

@@ -0,0 +1,58 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx;
+
+/**
+ * Provides a mechanism for receiving push-based notifications.
+ * <p>
+ * After an Observer calls an {@link Observable}'s <code>Observable.subscribe</code> method, the {@link Observable} calls the
+ * Observer's <code>onNext</code> method to provide notifications. A well-behaved {@link Observable} will call an Observer's
+ * <code>onCompleted</code> closure exactly once or the Observer's <code>onError</code> closure exactly once.
+ * <p>
+ * For more information see the <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>
+ * 
+ * @param <T>
+ */
+public interface Observer<T> {
+
+    /**
+     * Notifies the Observer that the {@link Observable} has finished sending push-based notifications.
+     * <p>
+     * The {@link Observable} will not call this closure if it calls <code>onError</code>.
+     */
+    public abstract void onCompleted();
+
+    /**
+     * Notifies the Observer that the {@link Observable} has experienced an error condition.
+     * <p>
+     * If the {@link Observable} calls this closure, it will not thereafter call <code>onNext</code> or <code>onCompleted</code>.
+     * 
+     * @param e
+     */
+    public abstract void onError(Throwable e);
+
+    /**
+     * Provides the Observer with new data.
+     * <p>
+     * The {@link Observable} calls this closure 1 or more times, unless it calls <code>onError</code> in which case this closure may never be called.
+     * <p>
+     * The {@link Observable} will not call this closure again after it calls either <code>onCompleted</code> or <code>onError</code>.
+     * 
+     * @param args
+     */
+    public abstract void onNext(T t);
+
+}

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

@@ -20,15 +20,15 @@
 /**
  * Provides a mechanism for receiving push-based notifications.
  * <p>
- * After an Observer calls an {@link Observable}'s <code>Observable.subscribe</code> method, the {@link Observable} calls the 
- * Observer's <code>onNext</code> method to provide notifications. A well-behaved {@link Observable} will call an Observer's 
+ * After an Observer calls an {@link Observable}'s <code>Observable.subscribe</code> method, the {@link Observable} calls the
+ * Observer's <code>onNext</code> method to provide notifications. A well-behaved {@link Observable} will call an Observer's
  * <code>onCompleted</code> closure exactly once or the Observer's <code>onError</code> closure exactly once.
  * <p>
  * For more information see the <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>
  * 
  * @param <T>
  */
-public abstract class Subscriber<T> implements Subscription {
+public abstract class Subscriber<T> implements Observer<T>, Subscription {
 
     private final CompositeSubscription cs;
 
@@ -47,32 +47,26 @@ protected Subscriber(Subscriber<?> op) {
         this(op.cs);
     }
 
-    /**
-     * Notifies the Observer that the {@link Observable} has finished sending push-based notifications.
-     * <p>
-     * The {@link Observable} will not call this closure if it calls <code>onError</code>.
-     */
-    public abstract void onCompleted();
+    public static <T> Subscriber<T> from(final Observer<? super T> o) {
+        return new Subscriber<T>() {
 
-    /**
-     * Notifies the Observer that the {@link Observable} has experienced an error condition.
-     * <p>
-     * If the {@link Observable} calls this closure, it will not thereafter call <code>onNext</code> or <code>onCompleted</code>.
-     * 
-     * @param e
-     */
-    public abstract void onError(Throwable e);
+            @Override
+            public void onCompleted() {
+                o.onCompleted();
+            }
 
-    /**
-     * Provides the Observer with new data.
-     * <p>
-     * The {@link Observable} calls this closure 1 or more times, unless it calls <code>onError</code> in which case this closure may never be called.
-     * <p>
-     * The {@link Observable} will not call this closure again after it calls either <code>onCompleted</code> or <code>onError</code>.
-     * 
-     * @param args
-     */
-    public abstract void onNext(T t);
+            @Override
+            public void onError(Throwable e) {
+                o.onError(e);
+            }
+
+            @Override
+            public void onNext(T t) {
+                o.onNext(t);
+            }
+
+        };
+    }
 
     /**
      * Used to register an unsubscribe callback.

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

@@ -24,7 +24,7 @@
 import rx.Notification;
 import rx.Observable;
 import rx.Subscriber;
-import rx.observers.SafeObserver;
+import rx.observers.SafeSubscriber;
 import rx.operators.SafeObservableSubscription;
 import rx.util.functions.Action1;
 
@@ -38,14 +38,14 @@ public final class JoinObserver1<T> extends Subscriber<Notification<T>> implemen
     private final List<ActivePlan0> activePlans;
     private final Queue<Notification<T>> queue;
     private final AtomicBoolean subscribed = new AtomicBoolean(false);
-    private final SafeObserver<Notification<T>> safeObserver;
+    private final SafeSubscriber<Notification<T>> safeObserver;
 
     public JoinObserver1(Observable<T> source, Action1<Throwable> onError) {
         this.source = source;
         this.onError = onError;
         queue = new LinkedList<Notification<T>>();
         activePlans = new ArrayList<ActivePlan0>();
-        safeObserver = new SafeObserver<Notification<T>>(new InnerObserver());
+        safeObserver = new SafeSubscriber<Notification<T>>(new InnerObserver());
         // add this subscription so it gets unsubscribed when the parent does
         add(safeObserver);
     }

rxjava-core/src/main/java/rx/joins/JoinObserver1.java

@@ -23,7 +23,7 @@
 import rx.Observable;
 import rx.Subscriber;
 import rx.Subscription;
-import rx.observers.SafeObserver;
+import rx.observers.SafeSubscriber;
 import rx.operators.OperationLatest;
 import rx.operators.OperationMostRecent;
 import rx.operators.OperationNext;
@@ -73,7 +73,7 @@ public static <T> BlockingObservable<T> from(final Observable<? extends T> o) {
      * "Guideline 6.4: Protect calls to user code from within an operator"
      */
     private Subscription protectivelyWrapAndSubscribe(Subscriber<? super T> observer) {
-        return o.subscribe(new SafeObserver<T>(observer));
+        return o.subscribe(new SafeSubscriber<T>(observer));
     }
 
     /**

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -1,11 +1,12 @@
 package rx.observers;
 
-import rx.Subscriber;
+import rx.Observer;
+
 
 /**
  * Observer that does nothing... including swallowing errors.
  */
-public class EmptyObserver<T> extends Subscriber<T> {
+public class EmptyObserver<T> implements Observer<T> {
 
     @Override
     public void onCompleted() {

rxjava-core/src/main/java/rx/observers/EmptyObserver.java

@@ -1,6 +1,6 @@
 package rx.observers;
 
-import rx.Subscriber;
+import rx.Observer;
 import rx.util.OnErrorNotImplementedException;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
@@ -10,8 +10,8 @@ public class Observers {
     /**
      * Create an empty Observer that ignores all events.
      */
-    public static final <T> Subscriber<T> create() {
-        return new Subscriber<T>() {
+    public static final <T> Observer<T> create() {
+        return new Observer<T>() {
 
             @Override
             public final void onCompleted() {
@@ -34,12 +34,12 @@ public final void onNext(T args) {
     /**
      * Create an Observer that receives `onNext` and ignores `onError` and `onCompleted`.
      */
-    public static final <T> Subscriber<T> create(final Action1<? super T> onNext) {
+    public static final <T> Observer<T> create(final Action1<? super T> onNext) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
 
-        return new Subscriber<T>() {
+        return new Observer<T>() {
 
             @Override
             public final void onCompleted() {
@@ -63,15 +63,15 @@ public final void onNext(T args) {
      * Create an Observer that receives `onNext` and `onError` and ignores `onCompleted`.
      * 
      */
-    public static final <T> Subscriber<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError) {
+    public static final <T> Observer<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
         if (onError == null) {
             throw new IllegalArgumentException("onError can not be null");
         }
 
-        return new Subscriber<T>() {
+        return new Observer<T>() {
 
             @Override
             public final void onCompleted() {
@@ -95,7 +95,7 @@ public final void onNext(T args) {
      * Create an Observer that receives `onNext`, `onError` and `onCompleted`.
      * 
      */
-    public static final <T> Subscriber<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
+    public static final <T> Observer<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
@@ -106,7 +106,7 @@ public static final <T> Subscriber<T> create(final Action1<? super T> onNext, fi
             throw new IllegalArgumentException("onComplete can not be null");
         }
 
-        return new Subscriber<T>() {
+        return new Observer<T>() {
 
             @Override
             public final void onCompleted() {

rxjava-core/src/main/java/rx/observers/Observers.java

@@ -58,12 +58,12 @@
  * 
  * @param <T>
  */
-public class SafeObserver<T> extends Subscriber<T> {
+public class SafeSubscriber<T> extends Subscriber<T> {
 
     private final Subscriber<? super T> actual;
     private final AtomicBoolean isFinished = new AtomicBoolean(false);
 
-    public SafeObserver(Subscriber<? super T> actual) {
+    public SafeSubscriber(Subscriber<? super T> actual) {
         super(actual);
         this.actual = actual;
     }

rxjava-core/src/main/java/rx/observers/SafeObserver.java renamed to rxjava-core/src/main/java/rx/observers/SafeSubscriber.java

@@ -15,6 +15,7 @@
  */
 package rx.observers;
 
+import rx.Observer;
 import rx.Subscriber;
 import rx.operators.SafeObservableSubscription;
 
@@ -30,7 +31,7 @@
  * 
  * @param <T>
  */
-public final class SynchronizedObserver<T> extends Subscriber<T> {
+public final class SynchronizedObserver<T> implements Observer<T> {
 
     /**
      * Intrinsic synchronized locking with double-check short-circuiting was chosen after testing several other implementations.
@@ -52,25 +53,31 @@ public final class SynchronizedObserver<T> extends Subscriber<T> {
     private volatile boolean finished = false;
     private volatile Object lock;
 
-    public SynchronizedObserver(Subscriber<? super T> Observer, SafeObservableSubscription subscription) {
-        this.observer = Observer;
+    public SynchronizedObserver(Subscriber<? super T> subscriber, SafeObservableSubscription subscription) {
+        this.observer = subscriber;
         this.subscription = subscription;
         this.lock = this;
     }
 
-    public SynchronizedObserver(Subscriber<? super T> Observer, SafeObservableSubscription subscription, Object lock) {
-        this.observer = Observer;
+    public SynchronizedObserver(Subscriber<? super T> subscriber, SafeObservableSubscription subscription, Object lock) {
+        this.observer = subscriber;
         this.subscription = subscription;
         this.lock = lock;
     }
+    
+    public SynchronizedObserver(Subscriber<? super T> subscriber, Object lock) {
+        this.observer = subscriber;
+        this.subscription = new SafeObservableSubscription(subscriber);
+        this.lock = lock;
+    }
 
     /**
      * Used when synchronizing an Observer without access to the subscription.
      * 
      * @param Observer
      */
-    public SynchronizedObserver(Subscriber<? super T> Observer) {
-        this(Observer, new SafeObservableSubscription());
+    public SynchronizedObserver(Subscriber<? super T> subscriber) {
+        this(subscriber, new SafeObservableSubscription());
     }
 
     public void onNext(T arg) {

rxjava-core/src/main/java/rx/observers/SynchronizedObserver.java

@@ -0,0 +1,69 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.observers;
+
+import rx.Observer;
+import rx.Subscriber;
+import rx.operators.SafeObservableSubscription;
+
+/**
+ * A thread-safe Observer for transitioning states in operators.
+ * <p>
+ * Execution rules are:
+ * <ul>
+ * <li>Allow only single-threaded, synchronous, ordered execution of onNext, onCompleted, onError</li>
+ * <li>Once an onComplete or onError are performed, no further calls can be executed</li>
+ * <li>If unsubscribe is called, this means we call completed() and don't allow any further onNext calls.</li>
+ * </ul>
+ * 
+ * @param <T>
+ */
+public final class SynchronizedSubscriber<T> extends Subscriber<T> {
+
+    private final Observer<? super T> observer;
+
+    public SynchronizedSubscriber(Subscriber<? super T> subscriber, Object lock) {
+        super(subscriber);
+        SafeObservableSubscription s = new SafeObservableSubscription();
+        subscriber.add(s);
+        this.observer = new SynchronizedObserver<T>(subscriber, s, lock);
+    }
+
+    /**
+     * Used when synchronizing an Subscriber without access to the subscription.
+     * 
+     * @param Observer
+     */
+    public SynchronizedSubscriber(Subscriber<? super T> subscriber) {
+        this(subscriber, new Object());
+    }
+
+    @Override
+    public void onCompleted() {
+        observer.onCompleted();
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        observer.onError(e);
+    }
+
+    @Override
+    public void onNext(T t) {
+        observer.onNext(t);
+    }
+
+}