Fixed 1 Unit Test - Linked Observer/Subscription where seemingly obvious to do so.

benjchristensen · benjchristensen · commit 0ac1e6c18596 · 2014-01-24T13:07:48.000-08:00
diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
@@ -213,7 +213,7 @@ public final static <T> Observable<T> create(final OnSubscribeFunc<T> f) {
             @Override
             public void call(Observer<? super T> observer) {
                 Subscription s = f.onSubscribe(observer);
-                if (s != null) {
+                if (s != null && s != observer) {
                     observer.add(s);
                 }
             }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationAny.java b/rxjava-core/src/main/java/rx/operators/OperationAny.java
@@ -80,8 +80,7 @@ private Any(Observable<? extends T> source, Func1<? super T, Boolean> predicate,
 
         @Override
         public Subscription onSubscribe(final Observer<? super Boolean> observer) {
-            final SafeObservableSubscription subscription = new SafeObservableSubscription();
-            return subscription.wrap(source.subscribe(new Observer<T>() {
+            return source.subscribe(new Observer<T>(observer) {
 
                 private final AtomicBoolean hasEmitted = new AtomicBoolean(false);
 
@@ -93,16 +92,12 @@ public void onNext(T value) {
                                     && hasEmitted.getAndSet(true) == false) {
                                 observer.onNext(!returnOnEmpty);
                                 observer.onCompleted();
-                                // this will work if the sequence is asynchronous, it
-                                // will have no effect on a synchronous observable
-                                subscription.unsubscribe();
+                                unsubscribe();
                             }
                         }
                     } catch (Throwable ex) {
                         observer.onError(ex);
-                        // this will work if the sequence is asynchronous, it
-                        // will have no effect on a synchronous observable
-                        subscription.unsubscribe();
+                        unsubscribe();
                     }
 
                 }
@@ -119,7 +114,7 @@ public void onCompleted() {
                         observer.onCompleted();
                     }
                 }
-            }));
+            });
         }
 
     }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDefaultIfEmpty.java b/rxjava-core/src/main/java/rx/operators/OperationDefaultIfEmpty.java
@@ -55,8 +55,7 @@ private DefaultIfEmpty(Observable<? extends T> source, T defaultValue) {
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            final SafeObservableSubscription subscription = new SafeObservableSubscription();
-            return subscription.wrap(source.subscribe(new Observer<T>() {
+            return source.subscribe(new Observer<T>(observer) {
 
                 private volatile boolean hasEmitted = false;
 
@@ -67,9 +66,7 @@ public void onNext(T value) {
                         observer.onNext(value);
                     } catch (Throwable ex) {
                         observer.onError(ex);
-                        // this will work if the sequence is asynchronous, it
-                        // will have no effect on a synchronous observable
-                        subscription.unsubscribe();
+                        unsubscribe();
                     }
                 }
 
@@ -87,7 +84,7 @@ public void onCompleted() {
                         observer.onCompleted();
                     }
                 }
-            }));
+            });
         }
     }
 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDematerialize.java b/rxjava-core/src/main/java/rx/operators/OperationDematerialize.java
@@ -54,7 +54,7 @@ public DematerializeObservable(Observable<? extends Notification<? extends T>> s
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            return sequence.subscribe(new Observer<Notification<? extends T>>() {
+            return sequence.subscribe(new Observer<Notification<? extends T>>(observer) {
                 @Override
                 public void onCompleted() {
                 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDistinct.java b/rxjava-core/src/main/java/rx/operators/OperationDistinct.java
@@ -97,7 +97,7 @@ private Distinct(Observable<? extends T> source, Func1<? super T, ? extends U> k
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            final Subscription sourceSub = source.subscribe(new Observer<T>() {
+            return source.subscribe(new Observer<T>(observer) {
                 private final Set<U> emittedKeys = new HashSet<U>();
 
                 @Override
@@ -119,13 +119,6 @@ public void onNext(T next) {
                     }
                 }
             });
-
-            return Subscriptions.create(new Action0() {
-                @Override
-                public void call() {
-                    sourceSub.unsubscribe();
-                }
-            });
         }
     }
 
@@ -142,7 +135,7 @@ private DistinctWithComparator(Observable<? extends T> source, Func1<? super T,
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            final Subscription sourceSub = source.subscribe(new Observer<T>() {
+            return source.subscribe(new Observer<T>(observer) {
 
                 // due to the totally arbitrary equality comparator, we can't use anything more efficient than lists here 
                 private final List<U> emittedKeys = new ArrayList<U>();
@@ -175,13 +168,6 @@ private boolean alreadyEmitted(U newKey) {
                     return false;
                 }
             });
-
-            return Subscriptions.create(new Action0() {
-                @Override
-                public void call() {
-                    sourceSub.unsubscribe();
-                }
-            });
         }
     }
 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDistinctUntilChanged.java b/rxjava-core/src/main/java/rx/operators/OperationDistinctUntilChanged.java
@@ -108,7 +108,7 @@ private DistinctUntilChanged(Observable<? extends T> source, Func1<? super T, ?
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            final Subscription sourceSub = source.subscribe(new Observer<T>() {
+            return source.subscribe(new Observer<T>(observer) {
                 private U lastEmittedKey;
                 private boolean hasEmitted;
 
@@ -135,13 +135,6 @@ public void onNext(T next) {
                     }
                 }
             });
-
-            return Subscriptions.create(new Action0() {
-                @Override
-                public void call() {
-                    sourceSub.unsubscribe();
-                }
-            });
         }
     }
 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDoOnEach.java b/rxjava-core/src/main/java/rx/operators/OperationDoOnEach.java
@@ -40,7 +40,7 @@ public DoOnEachObservable(Observable<? extends T> sequence, Observer<? super T>
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            return sequence.subscribe(new Observer<T>() {
+            return sequence.subscribe(new Observer<T>(observer) {
                 @Override
                 public void onCompleted() {
                     try {
diff --git a/rxjava-core/src/main/java/rx/operators/OperationFilter.java b/rxjava-core/src/main/java/rx/operators/OperationFilter.java
@@ -43,17 +43,15 @@ public Filter(Observable<? extends T> that, Func1<? super T, Boolean> predicate)
         }
 
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            final SafeObservableSubscription subscription = new SafeObservableSubscription();
-            return subscription.wrap(that.subscribe(new Observer<T>() {
+            return that.subscribe(new Observer<T>(observer) {
                 public void onNext(T value) {
                     try {
                         if (predicate.call(value)) {
                             observer.onNext(value);
                         }
                     } catch (Throwable ex) {
                         observer.onError(ex);
-                        // this will work if the sequence is asynchronous, it will have no effect on a synchronous observable
-                        subscription.unsubscribe();
+                        unsubscribe();
                     }
                 }
 
@@ -64,7 +62,7 @@ public void onError(Throwable ex) {
                 public void onCompleted() {
                     observer.onCompleted();
                 }
-            }));
+            });
         }
 
     }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationMaterialize.java b/rxjava-core/src/main/java/rx/operators/OperationMaterialize.java
@@ -54,23 +54,23 @@ public MaterializeObservable(Observable<? extends T> sequence) {
 
         @Override
         public Subscription onSubscribe(final Observer<? super Notification<T>> observer) {
-            return sequence.subscribe(new Observer<T>() {
+            return sequence.subscribe(new Observer<T>(observer) {
 
                 @Override
                 public void onCompleted() {
-                    observer.onNext(new Notification<T>());
+                    observer.onNext(Notification.<T>createOnCompleted());
                     observer.onCompleted();
                 }
 
                 @Override
                 public void onError(Throwable e) {
-                    observer.onNext(new Notification<T>(e));
+                    observer.onNext(Notification.<T>createOnError(e));
                     observer.onCompleted();
                 }
 
                 @Override
                 public void onNext(T value) {
-                    observer.onNext(new Notification<T>(value));
+                    observer.onNext(Notification.createOnNext(value));
                 }
 
             });
diff --git a/rxjava-core/src/main/java/rx/operators/OperationRepeat.java b/rxjava-core/src/main/java/rx/operators/OperationRepeat.java
@@ -47,7 +47,7 @@ public Subscription onSubscribe(final Observer<? super T> observer) {
         compositeSubscription.add(scheduler.schedule(new Action1<Action0>() {
             @Override
             public void call(final Action0 self) {
-                innerSubscription.set(source.subscribe(new Observer<T>() {
+                innerSubscription.set(source.subscribe(new Observer<T>(observer) {
 
                     @Override
                     public void onCompleted() {
diff --git a/rxjava-core/src/main/java/rx/operators/OperationRetry.java b/rxjava-core/src/main/java/rx/operators/OperationRetry.java
@@ -93,7 +93,7 @@ public void onError(Throwable e) {
                             if ((retryCount == INFINITE_RETRY || attempts.get() <= retryCount) && !subscription.isUnsubscribed()) {
                                 // retry again
                                 // add the new subscription and schedule a retry recursively
-                                rescursiveSubscription.setSubscription(scheduler.schedule(rescursiveSubscription, attemptSubscription(observer)));
+                                rescursiveSubscription.set(scheduler.schedule(rescursiveSubscription, attemptSubscription(observer)));
                             } else {
                                 // give up and pass the failure
                                 observer.onError(e);
diff --git a/rxjava-core/src/main/java/rx/operators/OperationSample.java b/rxjava-core/src/main/java/rx/operators/OperationSample.java
@@ -107,13 +107,15 @@ public void onNext(T value) {
                 }
             });
 
-            return Subscriptions.create(new Action0() {
+            observer.add(Subscriptions.create(new Action0() {
                 @Override
                 public void call() {
                     clockSubscription.unsubscribe();
                     sourceSubscription.unsubscribe();
                 }
-            });
+            }));
+            
+            return observer;
         }
     }
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationScan.java b/rxjava-core/src/main/java/rx/operators/OperationScan.java
@@ -81,7 +81,7 @@ private AccuWithoutInitialValue(Observable<? extends T> sequence, Func2<T, T, T>
 
         @Override
         public Subscription onSubscribe(final Observer<? super T> observer) {
-            return sequence.subscribe(new Observer<T>() {
+            return sequence.subscribe(new Observer<T>(observer) {
 
                 // has to be synchronized so that the initial value is always sent only once.
                 @Override
@@ -132,6 +132,7 @@ private static class AccumulatingObserver<T, R> extends Observer<T> {
         private R acc;
 
         private AccumulatingObserver(Observer<? super R> observer, R initialValue, Func2<R, ? super T, R> accumulator) {
+            super(observer);
             this.observer = observer;
             this.accumulatorFunction = accumulator;
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationSingle.java b/rxjava-core/src/main/java/rx/operators/OperationSingle.java
@@ -44,8 +44,7 @@ private static <T> OnSubscribeFunc<T> single(
 
             @Override
             public Subscription onSubscribe(final Observer<? super T> observer) {
-                final SafeObservableSubscription subscription = new SafeObservableSubscription();
-                subscription.wrap(source.subscribe(new Observer<T>() {
+                return source.subscribe(new Observer<T>(observer) {
 
                     private T value;
                     private boolean isEmpty = true;
@@ -85,11 +84,10 @@ public void onNext(T value) {
                             hasTooManyElemenets = true;
                             observer.onError(new IllegalArgumentException(
                                     "Sequence contains too many elements"));
-                            subscription.unsubscribe();
+                            unsubscribe();
                         }
                     }
-                }));
-                return subscription;
+                });
             }
         };
     }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationSkipLast.java b/rxjava-core/src/main/java/rx/operators/OperationSkipLast.java
@@ -74,8 +74,7 @@ public Subscription onSubscribe(final Observer<? super T> observer) {
                 throw new IndexOutOfBoundsException(
                         "count could not be negative");
             }
-            final SafeObservableSubscription subscription = new SafeObservableSubscription();
-            return subscription.wrap(source.subscribe(new Observer<T>() {
+            return source.subscribe(new Observer<T>(observer) {
 
                 private final ReentrantLock lock = new ReentrantLock();
 
@@ -104,7 +103,7 @@ public void onNext(T value) {
                             observer.onNext(value);
                         } catch (Throwable ex) {
                             observer.onError(ex);
-                            subscription.unsubscribe();
+                            unsubscribe();
                         }
                         return;
                     }
@@ -120,13 +119,13 @@ public void onNext(T value) {
                         }
                     } catch (Throwable ex) {
                         observer.onError(ex);
-                        subscription.unsubscribe();
+                        unsubscribe();
                     } finally {
                         lock.unlock();
                     }
                 }
 
-            }));
+            });
         }
     }
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationZip.java b/rxjava-core/src/main/java/rx/operators/OperationZip.java
@@ -150,7 +150,7 @@ public Subscription onSubscribe(final Observer<? super U> observer) {
 
             final List<ItemObserver<T>> all = new ArrayList<ItemObserver<T>>();
 
-            Observer<List<T>> o2 = new Observer<List<T>>() {
+            Observer<List<T>> o2 = new Observer<List<T>>(observer) {
                 boolean done;
                 @Override
                 public void onCompleted() {

Original file line number	Diff line number	Diff line change
`@@ -213,7 +213,7 @@ public final static <T> Observable<T> create(final OnSubscribeFunc<T> f) {`
`213`	`213`	`@Override`
`214`	`214`	`public void call(Observer<? super T> observer) {`
`215`	`215`	`Subscription s = f.onSubscribe(observer);`
`216`		`- if (s != null) {`
	`216`	`+ if (s != null && s != observer) {`
`217`	`217`	`observer.add(s);`
`218`	`218`	`}`
`219`	`219`	`}`
Original file line number	Diff line number	Diff line change
`@@ -80,8 +80,7 @@ private Any(Observable<? extends T> source, Func1<? super T, Boolean> predicate,`
`80`	`80`
`81`	`81`	`@Override`
`82`	`82`	`public Subscription onSubscribe(final Observer<? super Boolean> observer) {`
`83`		`- final SafeObservableSubscription subscription = new SafeObservableSubscription();`
`84`		`- return subscription.wrap(source.subscribe(new Observer<T>() {`
	`83`	`+ return source.subscribe(new Observer<T>(observer) {`
`85`	`84`
`86`	`85`	`private final AtomicBoolean hasEmitted = new AtomicBoolean(false);`
`87`	`86`
`@@ -93,16 +92,12 @@ public void onNext(T value) {`
`93`	`92`	`&& hasEmitted.getAndSet(true) == false) {`
`94`	`93`	`observer.onNext(!returnOnEmpty);`
`95`	`94`	`observer.onCompleted();`
`96`		`- // this will work if the sequence is asynchronous, it`
`97`		`- // will have no effect on a synchronous observable`
`98`		`- subscription.unsubscribe();`
	`95`	`+ unsubscribe();`
`99`	`96`	`}`
`100`	`97`	`}`
`101`	`98`	`} catch (Throwable ex) {`
`102`	`99`	`observer.onError(ex);`
`103`		`- // this will work if the sequence is asynchronous, it`
`104`		`- // will have no effect on a synchronous observable`
`105`		`- subscription.unsubscribe();`
	`100`	`+ unsubscribe();`
`106`	`101`	`}`
`107`	`102`
`108`	`103`	`}`
`@@ -119,7 +114,7 @@ public void onCompleted() {`
`119`	`114`	`observer.onCompleted();`
`120`	`115`	`}`
`121`	`116`	`}`
`122`		`- }));`
	`117`	`+ });`
`123`	`118`	`}`
`124`	`119`
`125`	`120`	`}`
Original file line number	Diff line number	Diff line change
`@@ -55,8 +55,7 @@ private DefaultIfEmpty(Observable<? extends T> source, T defaultValue) {`
`55`	`55`
`56`	`56`	`@Override`
`57`	`57`	`public Subscription onSubscribe(final Observer<? super T> observer) {`
`58`		`- final SafeObservableSubscription subscription = new SafeObservableSubscription();`
`59`		`- return subscription.wrap(source.subscribe(new Observer<T>() {`
	`58`	`+ return source.subscribe(new Observer<T>(observer) {`
`60`	`59`
`61`	`60`	`private volatile boolean hasEmitted = false;`
`62`	`61`
`@@ -67,9 +66,7 @@ public void onNext(T value) {`
`67`	`66`	`observer.onNext(value);`
`68`	`67`	`} catch (Throwable ex) {`
`69`	`68`	`observer.onError(ex);`
`70`		`- // this will work if the sequence is asynchronous, it`
`71`		`- // will have no effect on a synchronous observable`
`72`		`- subscription.unsubscribe();`
	`69`	`+ unsubscribe();`
`73`	`70`	`}`
`74`	`71`	`}`
`75`	`72`
`@@ -87,7 +84,7 @@ public void onCompleted() {`
`87`	`84`	`observer.onCompleted();`
`88`	`85`	`}`
`89`	`86`	`}`
`90`		`- }));`
	`87`	`+ });`
`91`	`88`	`}`
`92`	`89`	`}`
`93`	`90`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ public DematerializeObservable(Observable<? extends Notification<? extends T>> s`
`54`	`54`
`55`	`55`	`@Override`
`56`	`56`	`public Subscription onSubscribe(final Observer<? super T> observer) {`
`57`		`- return sequence.subscribe(new Observer<Notification<? extends T>>() {`
	`57`	`+ return sequence.subscribe(new Observer<Notification<? extends T>>(observer) {`
`58`	`58`	`@Override`
`59`	`59`	`public void onCompleted() {`
`60`	`60`	`}`
Original file line number	Diff line number	Diff line change
`@@ -97,7 +97,7 @@ private Distinct(Observable<? extends T> source, Func1<? super T, ? extends U> k`
`97`	`97`
`98`	`98`	`@Override`
`99`	`99`	`public Subscription onSubscribe(final Observer<? super T> observer) {`
`100`		`- final Subscription sourceSub = source.subscribe(new Observer<T>() {`
	`100`	`+ return source.subscribe(new Observer<T>(observer) {`
`101`	`101`	`private final Set<U> emittedKeys = new HashSet<U>();`
`102`	`102`
`103`	`103`	`@Override`
`@@ -119,13 +119,6 @@ public void onNext(T next) {`
`119`	`119`	`}`
`120`	`120`	`}`
`121`	`121`	`});`
`122`		`-`
`123`		`- return Subscriptions.create(new Action0() {`
`124`		`- @Override`
`125`		`- public void call() {`
`126`		`- sourceSub.unsubscribe();`
`127`		`- }`
`128`		`- });`
`129`	`122`	`}`
`130`	`123`	`}`
`131`	`124`
`@@ -142,7 +135,7 @@ private DistinctWithComparator(Observable<? extends T> source, Func1<? super T,`
`142`	`135`
`143`	`136`	`@Override`
`144`	`137`	`public Subscription onSubscribe(final Observer<? super T> observer) {`
`145`		`- final Subscription sourceSub = source.subscribe(new Observer<T>() {`
	`138`	`+ return source.subscribe(new Observer<T>(observer) {`
`146`	`139`
`147`	`140`	`// due to the totally arbitrary equality comparator, we can't use anything more efficient than lists here`
`148`	`141`	`private final List<U> emittedKeys = new ArrayList<U>();`
`@@ -175,13 +168,6 @@ private boolean alreadyEmitted(U newKey) {`
`175`	`168`	`return false;`
`176`	`169`	`}`
`177`	`170`	`});`
`178`		`-`
`179`		`- return Subscriptions.create(new Action0() {`
`180`		`- @Override`
`181`		`- public void call() {`
`182`		`- sourceSub.unsubscribe();`
`183`		`- }`
`184`		`- });`
`185`	`171`	`}`
`186`	`172`	`}`
`187`	`173`	`}`
Original file line number	Diff line number	Diff line change
`@@ -108,7 +108,7 @@ private DistinctUntilChanged(Observable<? extends T> source, Func1<? super T, ?`
`108`	`108`
`109`	`109`	`@Override`
`110`	`110`	`public Subscription onSubscribe(final Observer<? super T> observer) {`
`111`		`- final Subscription sourceSub = source.subscribe(new Observer<T>() {`
	`111`	`+ return source.subscribe(new Observer<T>(observer) {`
`112`	`112`	`private U lastEmittedKey;`
`113`	`113`	`private boolean hasEmitted;`
`114`	`114`
`@@ -135,13 +135,6 @@ public void onNext(T next) {`
`135`	`135`	`}`
`136`	`136`	`}`
`137`	`137`	`});`
`138`		`-`
`139`		`- return Subscriptions.create(new Action0() {`
`140`		`- @Override`
`141`		`- public void call() {`
`142`		`- sourceSub.unsubscribe();`
`143`		`- }`
`144`		`- });`
`145`	`138`	`}`
`146`	`139`	`}`
`147`	`140`	`}`
Original file line number	Diff line number	Diff line change
`@@ -43,17 +43,15 @@ public Filter(Observable<? extends T> that, Func1<? super T, Boolean> predicate)`
`43`	`43`	`}`
`44`	`44`
`45`	`45`	`public Subscription onSubscribe(final Observer<? super T> observer) {`
`46`		`- final SafeObservableSubscription subscription = new SafeObservableSubscription();`
`47`		`- return subscription.wrap(that.subscribe(new Observer<T>() {`
	`46`	`+ return that.subscribe(new Observer<T>(observer) {`
`48`	`47`	`public void onNext(T value) {`
`49`	`48`	`try {`
`50`	`49`	`if (predicate.call(value)) {`
`51`	`50`	`observer.onNext(value);`
`52`	`51`	`}`
`53`	`52`	`} catch (Throwable ex) {`
`54`	`53`	`observer.onError(ex);`
`55`		`- // this will work if the sequence is asynchronous, it will have no effect on a synchronous observable`
`56`		`- subscription.unsubscribe();`
	`54`	`+ unsubscribe();`
`57`	`55`	`}`
`58`	`56`	`}`
`59`	`57`
`@@ -64,7 +62,7 @@ public void onError(Throwable ex) {`
`64`	`62`	`public void onCompleted() {`
`65`	`63`	`observer.onCompleted();`
`66`	`64`	`}`
`67`		`- }));`
	`65`	`+ });`
`68`	`66`	`}`
`69`	`67`
`70`	`68`	`}`