Rename observer to subscriber and add a special logic to exit the loop

Author: zsxwing

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

@@ -21,7 +21,6 @@
 
 import rx.Observable;
 import rx.Observable.OnSubscribe;
-import rx.Observer;
 import rx.Subscriber;
 
 /**
@@ -121,12 +120,12 @@ private static final class AmbSubscriber<T> extends Subscriber<T> {
 
         private static final int NONE = -1;
 
-        private Observer<? super T> observer;
-        private int index;
-        private AtomicInteger choice;
+        private final Subscriber<? super T> subscriber;
+        private final int index;
+        private final AtomicInteger choice;
 
-        private AmbSubscriber(Subscriber<? super T> observer, int index, AtomicInteger choice) {
-            this.observer = observer;
+        private AmbSubscriber(Subscriber<? super T> subscriber, int index, AtomicInteger choice) {
+            this.subscriber = subscriber;
             this.choice = choice;
             this.index = index;
         }
@@ -137,7 +136,7 @@ public void onNext(T args) {
                 unsubscribe();
                 return;
             }
-            observer.onNext(args);
+            subscriber.onNext(args);
         }
 
         @Override
@@ -146,7 +145,7 @@ public void onCompleted() {
                 unsubscribe();
                 return;
             }
-            observer.onCompleted();
+            subscriber.onCompleted();
         }
 
         @Override
@@ -155,7 +154,7 @@ public void onError(Throwable e) {
                 unsubscribe();
                 return;
             }
-            observer.onError(e);
+            subscriber.onError(e);
         }
 
         private boolean isSelected() {
@@ -180,6 +179,10 @@ public void call(Subscriber<? super T> subscriber) {
             if (subscriber.isUnsubscribed()) {
                 break;
             }
+            if (choice.get() != AmbSubscriber.NONE) {
+                // Already choose someone, the rest Observables can be skipped.
+                break;
+            }
             AmbSubscriber<T> ambSubscriber = new AmbSubscriber<T>(subscriber, index, choice);
             subscriber.add(ambSubscriber);
             source.subscribe(ambSubscriber);