Fix OperationSwitch so that it does not onComplete() before inner and

outer subscription completes.

Author: ylecaillez

rxjava-core/src/main/java/rx/operators/OperationSwitch.java

@@ -34,6 +34,7 @@
 import rx.Observer;
 import rx.Subscription;
 import rx.concurrency.TestScheduler;
+import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.MultipleAssignmentSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
@@ -78,9 +79,15 @@ public Switch(Observable<? extends Observable<? extends T>> sequences) {
 
         @Override
         public Subscription onSubscribe(Observer<? super T> observer) {
-            SafeObservableSubscription subscription = new SafeObservableSubscription();
-            subscription.wrap(sequences.subscribe(new SwitchObserver<T>(observer, subscription)));
-            return subscription;
+            SafeObservableSubscription parent;
+            parent = new SafeObservableSubscription();
+
+            MultipleAssignmentSubscription child;
+            child = new MultipleAssignmentSubscription();
+
+            parent.wrap(sequences.subscribe(new SwitchObserver<T>(observer, parent, child)));
+
+            return new CompositeSubscription(parent, child);
         }
     }
 
@@ -89,24 +96,25 @@ private static class SwitchObserver<T> implements Observer<Observable<? extends
         private final Object                         gate;
         private final Observer<? super T>            observer;
         private final SafeObservableSubscription     parent;
-        private final MultipleAssignmentSubscription innerSubscription;
+        private final MultipleAssignmentSubscription child;
         private long                                 latest;
         private boolean                              stopped;
         private boolean                              hasLatest;
 
-        public SwitchObserver(Observer<? super T> observer, SafeObservableSubscription parent) {
+        public SwitchObserver(Observer<? super T> observer, SafeObservableSubscription parent,
+                MultipleAssignmentSubscription child) {
             this.observer = observer;
             this.parent = parent;
+            this.child = child;
             this.gate = new Object();
-            this.innerSubscription = new MultipleAssignmentSubscription();
         }
 
         @Override
         public void onNext(Observable<? extends T> args) {
             final long id;
             synchronized (gate) {
                 id = ++latest;
-                hasLatest = true;
+                this.hasLatest = true;
             }
 
             final SafeObservableSubscription sub;
@@ -116,7 +124,7 @@ public void onNext(Observable<? extends T> args) {
                 public void onNext(T args) {
                     synchronized (gate) {
                         if (latest == id) {
-                            observer.onNext(args);
+                            SwitchObserver.this.observer.onNext(args);
                         }
                     }
                 }
@@ -126,8 +134,8 @@ public void onError(Throwable e) {
                     synchronized (gate) {
                         sub.unsubscribe();
                         if (latest == id) {
-                            observer.onError(e);
-                            parent.unsubscribe();
+                            SwitchObserver.this.observer.onError(e);
+                            SwitchObserver.this.parent.unsubscribe();
                         }
                     }
                 }
@@ -137,39 +145,39 @@ public void onCompleted() {
                     synchronized (gate) {
                         sub.unsubscribe();
                         if (latest == id) {
-                            hasLatest = false;
+                            SwitchObserver.this.hasLatest = false;
                         }
 
                         if (stopped) {
-                            observer.onCompleted();
-                            parent.unsubscribe();
+                            SwitchObserver.this.observer.onCompleted();
+                            SwitchObserver.this.parent.unsubscribe();
                         }
 
                     }
                 }
 
             }));
 
-            innerSubscription.setSubscription(sub);
+            this.child.setSubscription(sub);
         }
 
         @Override
         public void onError(Throwable e) {
             synchronized (gate) {
-                observer.onError(e);
+                this.observer.onError(e);
             }
 
-            parent.unsubscribe();
+            this.parent.unsubscribe();
         }
 
         @Override
         public void onCompleted() {
             synchronized (gate) {
-                innerSubscription.unsubscribe();
-                stopped = true;
-                if (!hasLatest) {
-                    observer.onCompleted();
-                    parent.unsubscribe();
+                this.child.unsubscribe();
+                this.stopped = true;
+                if (!this.hasLatest) {
+                    this.observer.onCompleted();
+                    this.parent.unsubscribe();
                 }
             }
         }