Operator Tests Retrofitted to use TestObserver

- work around inability of Mockito to correctly mock an abstract class
- 15 of 590 tests still failing

Author: benjchristensen

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

@@ -104,7 +104,7 @@
 import rx.operators.OperatorMerge;
 import rx.operators.OperatorParallel;
 import rx.operators.OperatorTake;
-import rx.operators.OperatorTakeTimed;
+import rx.operators.OperationTakeTimed;
 import rx.operators.OperatorTimestamp;
 import rx.operators.OperatorToObservableList;
 import rx.operators.OperatorToObservableSortedList;
@@ -212,7 +212,10 @@ public final static <T> Observable<T> create(final OnSubscribeFunc<T> f) {
 
             @Override
             public void call(Observer<? super T> observer) {
-                f.onSubscribe(observer);
+                Subscription s = f.onSubscribe(observer);
+                if (s != null) {
+                    observer.add(s);
+                }
             }
 
         });
@@ -7134,7 +7137,7 @@ public final Observable<T> take(long time, TimeUnit unit) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#take">RxJava Wiki: take()</a>
      */
     public final Observable<T> take(long time, TimeUnit unit, Scheduler scheduler) {
-        return create(new OperatorTakeTimed.TakeTimed<T>(this, time, unit, scheduler));
+        return create(new OperationTakeTimed.TakeTimed<T>(this, time, unit, scheduler));
     }
 
     /**
@@ -8465,7 +8468,7 @@ private boolean isInternalImplementation(Object o) {
         } else {
             // we treat the following package as "internal" and don't wrap it
             Package p = o.getClass().getPackage(); // it can be null
-            Boolean isInternal = (p != null && p.getName().startsWith("rx.Observers"));
+            Boolean isInternal = (p != null && p.getName().startsWith("rx.operators"));
             internalClassMap.put(clazz, isInternal);
             return isInternal;
         }

rxjava-core/src/main/java/rx/operators/OperatorTakeTimed.java renamed to rxjava-core/src/main/java/rx/operators/OperationTakeTimed.java

@@ -38,7 +38,7 @@
  * subscribing Observer's <code>onNext</code> function a maximum of <code>num</code> times before
  * invoking <code>onCompleted</code>.
  */
-public final class OperatorTakeTimed {
+public final class OperationTakeTimed {
 
     //TODO this has not been migrated to use bind yet
 

rxjava-core/src/test/java/rx/operators/OperationAmbTest.java

@@ -29,6 +29,7 @@
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Subscription;
+import rx.observers.TestObserver;
 import rx.schedulers.TestScheduler;
 import rx.subscriptions.CompositeSubscription;
 import rx.util.functions.Action0;
@@ -90,7 +91,7 @@ public void testAmb() {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = (Observer<String>) mock(Observer.class);
-        o.subscribe(observer);
+        o.subscribe(new TestObserver<String>(observer));
 
         scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS);
 
@@ -119,7 +120,7 @@ public void testAmb2() {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = (Observer<String>) mock(Observer.class);
-        o.subscribe(observer);
+        o.subscribe(new TestObserver<String>(observer));
 
         scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS);
 
@@ -146,7 +147,7 @@ public void testAmb3() {
 
         @SuppressWarnings("unchecked")
         Observer<String> observer = (Observer<String>) mock(Observer.class);
-        o.subscribe(observer);
+        o.subscribe(new TestObserver<String>(observer));
 
         scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS);
         InOrder inOrder = inOrder(observer);

rxjava-core/src/test/java/rx/operators/OperationAverageTest.java

@@ -23,6 +23,7 @@
 
 import rx.Observable;
 import rx.Observer;
+import rx.observers.TestObserver;
 import rx.operators.OperationReduceTest.CustomException;
 import rx.util.functions.Func1;
 
@@ -281,7 +282,7 @@ public Integer call(String t1) {
 
         Observable<Integer> result = source.averageInteger(length);
         Observer<Object> o = mock(Observer.class);
-        result.subscribe(o);
+        result.subscribe(new TestObserver<Object>(o));
 
         testThrows(o, CustomException.class);
     }
@@ -298,7 +299,7 @@ public Long call(String t1) {
 
         Observable<Long> result = source.averageLong(length);
         Observer<Object> o = mock(Observer.class);
-        result.subscribe(o);
+        result.subscribe(new TestObserver<Object>(o));
 
         testThrows(o, CustomException.class);
     }
@@ -315,7 +316,7 @@ public Float call(String t1) {
 
         Observable<Float> result = source.averageFloat(length);
         Observer<Object> o = mock(Observer.class);
-        result.subscribe(o);
+        result.subscribe(new TestObserver<Object>(o));
 
         testThrows(o, CustomException.class);
     }
@@ -332,7 +333,7 @@ public Double call(String t1) {
 
         Observable<Double> result = source.averageDouble(length);
         Observer<Object> o = mock(Observer.class);
-        result.subscribe(o);
+        result.subscribe(new TestObserver<Object>(o));
 
         testThrows(o, CustomException.class);
     }

rxjava-core/src/test/java/rx/operators/OperationBufferTest.java

@@ -34,6 +34,7 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.observers.TestObserver;
 import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
@@ -166,7 +167,7 @@ public Subscription onSubscribe(Observer<? super String> observer) {
         });
 
         Observable<List<String>> buffered = Observable.create(buffer(source, 100, TimeUnit.MILLISECONDS, 2, scheduler));
-        buffered.subscribe(observer);
+        buffered.subscribe(new TestObserver<List<String>>(observer));
 
         InOrder inOrder = Mockito.inOrder(observer);
         scheduler.advanceTimeTo(100, TimeUnit.MILLISECONDS);
@@ -198,7 +199,7 @@ public Subscription onSubscribe(Observer<? super String> observer) {
         });
 
         Observable<List<String>> buffered = Observable.create(buffer(source, 100, TimeUnit.MILLISECONDS, scheduler));
-        buffered.subscribe(observer);
+        buffered.subscribe(new TestObserver<List<String>>(observer));
 
         InOrder inOrder = Mockito.inOrder(observer);
         scheduler.advanceTimeTo(101, TimeUnit.MILLISECONDS);
@@ -251,7 +252,7 @@ public Subscription onSubscribe(Observer<? super Object> observer) {
         };
 
         Observable<List<String>> buffered = Observable.create(buffer(source, openings, closer));
-        buffered.subscribe(observer);
+        buffered.subscribe(new TestObserver<List<String>>(observer));
 
         InOrder inOrder = Mockito.inOrder(observer);
         scheduler.advanceTimeTo(500, TimeUnit.MILLISECONDS);
@@ -292,7 +293,7 @@ public Subscription onSubscribe(Observer<? super Object> observer) {
         };
 
         Observable<List<String>> buffered = Observable.create(buffer(source, closer));
-        buffered.subscribe(observer);
+        buffered.subscribe(new TestObserver<List<String>>(observer));
 
         InOrder inOrder = Mockito.inOrder(observer);
         scheduler.advanceTimeTo(500, TimeUnit.MILLISECONDS);
@@ -370,7 +371,7 @@ public void testBufferStopsWhenUnsubscribed1() {
 
         Observer<List<Integer>> o = mock(Observer.class);
 
-        Subscription s = source.buffer(100, 200, TimeUnit.MILLISECONDS, scheduler).subscribe(o);
+        Subscription s = source.buffer(100, 200, TimeUnit.MILLISECONDS, scheduler).subscribe(new TestObserver<List<Integer>>(o));
 
         InOrder inOrder = Mockito.inOrder(o);
 
@@ -394,7 +395,7 @@ public void bufferWithBONormal1() {
         Observer<Object> o = mock(Observer.class);
         InOrder inOrder = Mockito.inOrder(o);
 
-        source.toObservable().buffer(boundary.toObservable()).subscribe(o);
+        source.toObservable().buffer(boundary.toObservable()).subscribe(new TestObserver<Object>(o));
 
         source.onNext(1);
         source.onNext(2);
@@ -430,7 +431,7 @@ public void bufferWithBOEmptyLastViaBoundary() {
         Observer<Object> o = mock(Observer.class);
         InOrder inOrder = Mockito.inOrder(o);
 
-        source.toObservable().buffer(boundary.toObservable()).subscribe(o);
+        source.toObservable().buffer(boundary.toObservable()).subscribe(new TestObserver<Object>(o));
 
         boundary.onCompleted();
 
@@ -450,7 +451,7 @@ public void bufferWithBOEmptyLastViaSource() {
         Observer<Object> o = mock(Observer.class);
         InOrder inOrder = Mockito.inOrder(o);
 
-        source.toObservable().buffer(boundary.toObservable()).subscribe(o);
+        source.toObservable().buffer(boundary.toObservable()).subscribe(new TestObserver<Object>(o));
 
         source.onCompleted();
 
@@ -470,7 +471,7 @@ public void bufferWithBOEmptyLastViaBoth() {
         Observer<Object> o = mock(Observer.class);
         InOrder inOrder = Mockito.inOrder(o);
 
-        source.toObservable().buffer(boundary.toObservable()).subscribe(o);
+        source.toObservable().buffer(boundary.toObservable()).subscribe(new TestObserver<Object>(o));
 
         source.onCompleted();
         boundary.onCompleted();
@@ -490,7 +491,7 @@ public void bufferWithBOSourceThrows() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        source.toObservable().buffer(boundary.toObservable()).subscribe(o);
+        source.toObservable().buffer(boundary.toObservable()).subscribe(new TestObserver<Object>(o));
         source.onNext(1);
         source.onError(new OperationReduceTest.CustomException());
 
@@ -507,7 +508,7 @@ public void bufferWithBOBoundaryThrows() {
         @SuppressWarnings("unchecked")
         Observer<Object> o = mock(Observer.class);
 
-        source.toObservable().buffer(boundary.toObservable()).subscribe(o);
+        source.toObservable().buffer(boundary.toObservable()).subscribe(new TestObserver<Object>(o));
 
         source.onNext(1);
         boundary.onError(new OperationReduceTest.CustomException());

rxjava-core/src/test/java/rx/operators/OperationCombineLatestTest.java

@@ -28,6 +28,7 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.observers.TestObserver;
 import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func2;
@@ -51,7 +52,7 @@ public String call(String v1, String v2) {
                 throw new RuntimeException("I don't work.");
             }
         }));
-        combined.subscribe(w);
+        combined.subscribe(new TestObserver<String>(w));
 
         w1.observer.onNext("first value of w1");
         w2.observer.onNext("first value of w2");
@@ -72,7 +73,7 @@ public void testCombineLatestDifferentLengthObservableSequences1() {
         TestObservable w3 = new TestObservable();
 
         Observable<String> combineLatestW = Observable.create(combineLatest(Observable.create(w1), Observable.create(w2), Observable.create(w3), getConcat3StringsCombineLatestFunction()));
-        combineLatestW.subscribe(w);
+        combineLatestW.subscribe(new TestObserver<String>(w));
 
         /* simulate sending data */
         // once for w1
@@ -110,7 +111,7 @@ public void testCombineLatestDifferentLengthObservableSequences2() {
         TestObservable w3 = new TestObservable();
 
         Observable<String> combineLatestW = Observable.create(combineLatest(Observable.create(w1), Observable.create(w2), Observable.create(w3), getConcat3StringsCombineLatestFunction()));
-        combineLatestW.subscribe(w);
+        combineLatestW.subscribe(new TestObserver<String>(w));
 
         /* simulate sending data */
         // 4 times for w1
@@ -146,7 +147,7 @@ public void testCombineLatestWithInterleavingSequences() {
         TestObservable w3 = new TestObservable();
 
         Observable<String> combineLatestW = Observable.create(combineLatest(Observable.create(w1), Observable.create(w2), Observable.create(w3), getConcat3StringsCombineLatestFunction()));
-        combineLatestW.subscribe(w);
+        combineLatestW.subscribe(new TestObserver<String>(w));
 
         /* simulate sending data */
         w1.observer.onNext("1a");
@@ -185,7 +186,7 @@ public void testCombineLatest2Types() {
         Observer<String> aObserver = mock(Observer.class);
 
         Observable<String> w = Observable.create(combineLatest(Observable.from("one", "two"), Observable.from(2, 3, 4), combineLatestFunction));
-        w.subscribe(aObserver);
+        w.subscribe(new TestObserver<String>(aObserver));
 
         verify(aObserver, never()).onError(any(Throwable.class));
         verify(aObserver, times(1)).onCompleted();
@@ -204,7 +205,7 @@ public void testCombineLatest3TypesA() {
         Observer<String> aObserver = mock(Observer.class);
 
         Observable<String> w = Observable.create(combineLatest(Observable.from("one", "two"), Observable.from(2), Observable.from(new int[] { 4, 5, 6 }), combineLatestFunction));
-        w.subscribe(aObserver);
+        w.subscribe(new TestObserver<String>(aObserver));
 
         verify(aObserver, never()).onError(any(Throwable.class));
         verify(aObserver, times(1)).onCompleted();
@@ -221,7 +222,7 @@ public void testCombineLatest3TypesB() {
         Observer<String> aObserver = mock(Observer.class);
 
         Observable<String> w = Observable.create(combineLatest(Observable.from("one"), Observable.from(2), Observable.from(new int[] { 4, 5, 6 }, new int[] { 7, 8 }), combineLatestFunction));
-        w.subscribe(aObserver);
+        w.subscribe(new TestObserver<String>(aObserver));
 
         verify(aObserver, never()).onError(any(Throwable.class));
         verify(aObserver, times(1)).onCompleted();
@@ -334,7 +335,7 @@ public void combineSimple() {
 
         Observer<Object> observer = mock(Observer.class);
 
-        source.subscribe(observer);
+        source.subscribe(new TestObserver<Object>(observer));
 
         InOrder inOrder = inOrder(observer);
 
@@ -380,8 +381,8 @@ public void combineMultipleObservers() {
         Observer<Object> observer1 = mock(Observer.class);
         Observer<Object> observer2 = mock(Observer.class);
 
-        source.subscribe(observer1);
-        source.subscribe(observer2);
+        source.subscribe(new TestObserver<Object>(observer1));
+        source.subscribe(new TestObserver<Object>(observer2));
 
         InOrder inOrder1 = inOrder(observer1);
         InOrder inOrder2 = inOrder(observer2);
@@ -435,7 +436,7 @@ public void testFirstNeverProduces() {
 
         Observer<Object> observer = mock(Observer.class);
 
-        source.subscribe(observer);
+        source.subscribe(new TestObserver<Object>(observer));
 
         InOrder inOrder = inOrder(observer);
 
@@ -458,7 +459,7 @@ public void testSecondNeverProduces() {
 
         Observer<Object> observer = mock(Observer.class);
 
-        source.subscribe(observer);
+        source.subscribe(new TestObserver<Object>(observer));
 
         InOrder inOrder = inOrder(observer);
 

rxjava-core/src/test/java/rx/operators/OperationConcatTest.java

@@ -33,6 +33,7 @@
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.observers.TestObserver;
 import rx.schedulers.TestScheduler;
 import rx.subscriptions.BooleanSubscription;
 
@@ -308,7 +309,7 @@ public void testConcatConcurrentWithInfinity() {
 
         Observable<String> concat = Observable.create(concatF);
 
-        concat.take(50).subscribe(aObserver);
+        concat.take(50).subscribe(new TestObserver<String>(aObserver));
 
         //Wait for the thread to start up.
         try {
@@ -404,7 +405,7 @@ public void testConcatUnsubscribe() {
 
         try {
             // Subscribe
-            s1.wrap(concat.subscribe(aObserver));
+            s1.wrap(concat.subscribe(new TestObserver<String>(aObserver)));
             //Block main thread to allow observable "w1" to complete and observable "w2" to call onNext once.
             callOnce.await();
             // Unsubcribe
@@ -447,7 +448,7 @@ public void testConcatUnsubscribeConcurrent() {
 
         Observable<String> concat = Observable.create(concatF);
 
-        Subscription s1 = concat.subscribe(aObserver);
+        Subscription s1 = concat.subscribe(new TestObserver<String>(aObserver));
 
         try {
             //Block main thread to allow observable "w1" to complete and observable "w2" to call onNext exactly once.
@@ -565,8 +566,8 @@ public void testMultipleObservers() {
         Observable<Long> timer = Observable.interval(500, TimeUnit.MILLISECONDS, s).take(2);
         Observable<Long> o = Observable.concat(timer, timer);
 
-        o.subscribe(o1);
-        o.subscribe(o2);
+        o.subscribe(new TestObserver<Object>(o1));
+        o.subscribe(new TestObserver<Object>(o2));
 
         InOrder inOrder1 = inOrder(o1);
         InOrder inOrder2 = inOrder(o2);