Moved unsubscribe out the gate

Author: zsxwing

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

@@ -96,8 +96,7 @@ public void onNext(Observable<? extends T> args) {
                 this.hasLatest = true;
             }
 
-            final SafeObservableSubscription sub;
-            sub = new SafeObservableSubscription();
+            final SafeObservableSubscription sub = new SafeObservableSubscription();
             sub.wrap(args.subscribe(new Observer<T>() {
                 @Override
                 public void onNext(T args) {
@@ -110,28 +109,36 @@ public void onNext(T args) {
 
                 @Override
                 public void onError(Throwable e) {
+                    sub.unsubscribe();
+                    SafeObservableSubscription s = null;
                     synchronized (gate) {
-                        sub.unsubscribe();
                         if (latest == id) {
                             SwitchObserver.this.observer.onError(e);
-                            SwitchObserver.this.parent.unsubscribe();
+                            s = SwitchObserver.this.parent;
                         }
                     }
+                    if(s != null) {
+                        s.unsubscribe();
+                    }
                 }
 
                 @Override
                 public void onCompleted() {
+                    sub.unsubscribe();
+                    SafeObservableSubscription s = null;
                     synchronized (gate) {
-                        sub.unsubscribe();
                         if (latest == id) {
                             SwitchObserver.this.hasLatest = false;
 
                             if (stopped) {
                                 SwitchObserver.this.observer.onCompleted();
-                                SwitchObserver.this.parent.unsubscribe();
+                                s = SwitchObserver.this.parent;
                             }
                         }
                     }
+                    if(s != null) {
+                        s.unsubscribe();
+                    }
                 }
 
             }));
@@ -150,13 +157,17 @@ public void onError(Throwable e) {
 
         @Override
         public void onCompleted() {
+            SafeObservableSubscription s = null;
             synchronized (gate) {
                 this.stopped = true;
                 if (!this.hasLatest) {
                     this.observer.onCompleted();
-                    this.parent.unsubscribe();
+                    s = this.parent;
                 }
             }
+            if(s != null) {
+                s.unsubscribe();
+            }
         }
 
     }