OperationJoinTest - fixing

Author: benjchristensen

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

@@ -37,7 +37,6 @@ public final class JoinObserver1<T> extends Observer<Notification<T>> implements
     private final Action1<Throwable> onError;
     private final List<ActivePlan0> activePlans;
     private final Queue<Notification<T>> queue;
-    //    private volatile boolean done;
     private final AtomicBoolean subscribed = new AtomicBoolean(false);
     private final SafeObserver<Notification<T>> safeObserver;
 
@@ -74,6 +73,30 @@ public void dequeue() {
         queue.remove();
     }
 
+
+    @Override
+    public void onNext(Notification<T> args) {
+        safeObserver.onNext(args);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        safeObserver.onError(e);
+    }
+
+    @Override
+    public void onCompleted() {
+        safeObserver.onCompleted();
+    }
+
+    void removeActivePlan(ActivePlan0 activePlan) {
+        activePlans.remove(activePlan);
+        if (activePlans.isEmpty()) {
+            unsubscribe();
+        }
+    }
+    
+    
     private final class InnerObserver extends Observer<Notification<T>> {
 
         @Override
@@ -105,26 +128,4 @@ public void onCompleted() {
         }
     }
 
-    @Override
-    public void onNext(Notification<T> args) {
-        safeObserver.onNext(args);
-    }
-
-    @Override
-    public void onError(Throwable e) {
-        safeObserver.onError(e);
-    }
-
-    @Override
-    public void onCompleted() {
-        safeObserver.onCompleted();
-    }
-
-    void removeActivePlan(ActivePlan0 activePlan) {
-        activePlans.remove(activePlan);
-        if (activePlans.isEmpty()) {
-            unsubscribe();
-        }
-    }
-
-}
+}

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

@@ -366,10 +366,13 @@ public void whenComplicated() {
         PublishSubject<Integer> zs = PublishSubject.create();
 
         Observable<Integer> m = Observable.when(
-                xs.toObservable().and(ys.toObservable()).then(add2),
-                xs.toObservable().and(zs.toObservable()).then(mul2),
-                ys.toObservable().and(zs.toObservable()).then(sub2)
+                xs.toObservable().and(ys.toObservable()).then(add2), // 1+4=5, 2+5=7, 3+6=9
+                xs.toObservable().and(zs.toObservable()).then(mul2), // 1*7=7, 2*8=16, 3*9=27
+                ys.toObservable().and(zs.toObservable()).then(sub2)  // 4-7=-3, 5-8=-3, 6-9=-3
                 );
+
+        // 5, 7, 9, 7, 16, 27, -3, -3, -3
+
         TestObserver<Object> to = new TestObserver<Object>(observer);
         m.subscribe(to);
 
@@ -394,7 +397,7 @@ public void whenComplicated() {
         zs.onCompleted(); // t == 300
 
         System.out.println("Events: " + to.getOnNextEvents());
-        
+
         InOrder inOrder = inOrder(observer);
         inOrder.verify(observer, times(1)).onNext(1 * 7);
         inOrder.verify(observer, times(1)).onNext(2 * 8);