Observer+Subscription -> Subscriber

Author: benjchristensen

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

@@ -142,7 +142,7 @@ public boolean isOnNext() {
         return getKind() == Kind.OnNext;
     }
 
-    public void accept(Observer<? super T> observer) {
+    public void accept(Subscriber<? super T> observer) {
         if (isOnNext()) {
             observer.onNext(getValue());
         } else if (isOnCompleted()) {

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

@@ -28,22 +28,22 @@
  * 
  * @param <T>
  */
-public abstract class Observer<T> implements Subscription {
+public abstract class Subscriber<T> implements Subscription {
 
     private final CompositeSubscription cs;
 
-    protected Observer(CompositeSubscription cs) {
+    protected Subscriber(CompositeSubscription cs) {
         if (cs == null) {
             throw new IllegalArgumentException("The CompositeSubscription can not be null");
         }
         this.cs = cs;
     }
 
-    protected Observer() {
+    protected Subscriber() {
         this(new CompositeSubscription());
     }
 
-    protected Observer(Observer<?> op) {
+    protected Subscriber(Subscriber<?> op) {
         this(op.cs);
     }
 

rxjava-core/src/main/java/rx/Observer.java renamed to rxjava-core/src/main/java/rx/Subscriber.java

@@ -18,7 +18,7 @@
 import rx.subscriptions.Subscriptions;
 
 /**
- * Subscription returns from {@link Observable#subscribe(Observer)} to allow unsubscribing.
+ * Subscription returns from {@link Observable#subscribe(Subscriber)} to allow unsubscribing.
  * <p>
  * See utilities in {@link Subscriptions} and implementations in the {@code rx.subscriptions} package.
  * <p>
@@ -27,9 +27,9 @@
 public interface Subscription {
 
     /**
-     * Stop receiving notifications on the {@link Observer} that was registered when this Subscription was received.
+     * Stop receiving notifications on the {@link Subscriber} that was registered when this Subscription was received.
      * <p>
-     * This allows unregistering an {@link Observer} before it has finished receiving all events (ie. before onCompleted is called).
+     * This allows unregistering an {@link Subscriber} before it has finished receiving all events (ie. before onCompleted is called).
      */
     public void unsubscribe();
 

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

@@ -23,15 +23,15 @@
 
 import rx.Notification;
 import rx.Observable;
-import rx.Observer;
+import rx.Subscriber;
 import rx.observers.SafeObserver;
 import rx.operators.SafeObservableSubscription;
 import rx.util.functions.Action1;
 
 /**
  * Default implementation of a join observer.
  */
-public final class JoinObserver1<T> extends Observer<Notification<T>> implements JoinObserver {
+public final class JoinObserver1<T> extends Subscriber<Notification<T>> implements JoinObserver {
     private Object gate;
     private final Observable<T> source;
     private final Action1<Throwable> onError;
@@ -97,7 +97,7 @@ void removeActivePlan(ActivePlan0 activePlan) {
     }
 
 
-    private final class InnerObserver extends Observer<Notification<T>> {
+    private final class InnerObserver extends Subscriber<Notification<T>> {
 
         @Override
         public void onNext(Notification<T> args) {

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

@@ -18,15 +18,15 @@
 import java.util.Map;
 
 import rx.Observable;
-import rx.Observer;
+import rx.Subscriber;
 import rx.util.functions.Action1;
 
 /**
  * Represents an execution plan for join patterns.
  */
 public abstract class Plan0<R> {
     public abstract ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions,
-            Observer<R> observer, Action1<ActivePlan0> deactivate);
+            Subscriber<R> observer, Action1<ActivePlan0> deactivate);
 
     @SuppressWarnings("unchecked")
     public static <T> JoinObserver1<T> createObserver(

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

@@ -18,7 +18,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
-import rx.Observer;
+import rx.Subscriber;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 import rx.util.functions.Actions;
@@ -45,7 +45,7 @@ public Func1<T1, R> selector() {
     }
 
     @Override
-    public ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions, final Observer<R> observer, final Action1<ActivePlan0> deactivate) {
+    public ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions, final Subscriber<R> observer, final Action1<ActivePlan0> deactivate) {
         Action1<Throwable> onError = Actions.onErrorFrom(observer);
 
         final JoinObserver1<T1> firstJoinObserver = createObserver(externalSubscriptions, expression.first(), onError);

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

@@ -18,7 +18,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
-import rx.Observer;
+import rx.Subscriber;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 import rx.util.functions.Action2;
@@ -39,7 +39,7 @@ public Plan2(Pattern2<T1, T2> expression, Func2<T1, T2, R> selector) {
 
     @Override
     public ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions,
-            final Observer<R> observer, final Action1<ActivePlan0> deactivate) {
+            final Subscriber<R> observer, final Action1<ActivePlan0> deactivate) {
         Action1<Throwable> onError = Actions.onErrorFrom(observer);
 
         final JoinObserver1<T1> firstJoinObserver = createObserver(externalSubscriptions, expression.first(), onError);

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

@@ -18,7 +18,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
-import rx.Observer;
+import rx.Subscriber;
 import rx.util.functions.Action0;
 import rx.util.functions.Action1;
 import rx.util.functions.Action3;
@@ -39,7 +39,7 @@ public Plan3(Pattern3<T1, T2, T3> expression, Func3<T1, T2, T3, R> selector) {
 
     @Override
     public ActivePlan0 activate(Map<Object, JoinObserver> externalSubscriptions,
-            final Observer<R> observer, final Action1<ActivePlan0> deactivate) {
+            final Subscriber<R> observer, final Action1<ActivePlan0> deactivate) {
         Action1<Throwable> onError = Actions.onErrorFrom(observer);
 
         final JoinObserver1<T1> firstJoinObserver = createObserver(externalSubscriptions, expression.first(), onError);

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

@@ -21,7 +21,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Observable;
-import rx.Observer;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.observers.SafeObserver;
 import rx.operators.OperationLatest;
@@ -66,13 +66,13 @@ public static <T> BlockingObservable<T> from(final Observable<? extends T> o) {
     }
 
     /**
-     * Used for protecting against errors being thrown from {@link Observer} implementations and ensuring onNext/onError/onCompleted contract
+     * Used for protecting against errors being thrown from {@link Subscriber} implementations and ensuring onNext/onError/onCompleted contract
      * compliance.
      * <p>
      * See https://github.com/Netflix/RxJava/issues/216 for discussion on
      * "Guideline 6.4: Protect calls to user code from within an operator"
      */
-    private Subscription protectivelyWrapAndSubscribe(Observer<? super T> observer) {
+    private Subscription protectivelyWrapAndSubscribe(Subscriber<? super T> observer) {
         return o.subscribe(new SafeObserver<T>(observer));
     }
 
@@ -82,8 +82,8 @@ private Subscription protectivelyWrapAndSubscribe(Observer<? super T> observer)
      * <p>
      * NOTE: This will block even if the Observable is asynchronous.
      * <p>
-     * This is similar to {@link Observable#subscribe(Observer)}, but it blocks.
-     * Because it blocks it does not need the {@link Observer#onCompleted()} or {@link Observer#onError(Throwable)} methods.
+     * This is similar to {@link Observable#subscribe(Subscriber)}, but it blocks.
+     * Because it blocks it does not need the {@link Subscriber#onCompleted()} or {@link Subscriber#onError(Throwable)} methods.
      * <p>
      * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/B.forEach.png">
      * 
@@ -103,7 +103,7 @@ public void forEach(final Action1<? super T> onNext) {
          * See https://github.com/Netflix/RxJava/issues/216 for discussion on
          * "Guideline 6.4: Protect calls to user code from within an operator"
          */
-        protectivelyWrapAndSubscribe(new Observer<T>() {
+        protectivelyWrapAndSubscribe(new Subscriber<T>() {
             @Override
             public void onCompleted() {
                 latch.countDown();