Merge pull request #245 from mairbek/multisubscribe

Moved state from Observable to Observer

Author: benjchristensen

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

@@ -23,7 +23,6 @@ private static class AllObservable<T> implements Func1<Observer<Boolean>, Subscr
         private final Observable<T> sequence;
         private final Func1<T, Boolean> predicate;
 
-        private final AtomicBoolean status = new AtomicBoolean(true);
         private final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
 
 
@@ -35,33 +34,45 @@ private AllObservable(Observable<T> sequence, Func1<T, Boolean> predicate) {
 
         @Override
         public Subscription call(final Observer<Boolean> observer) {
-            return subscription.wrap(sequence.subscribe(new Observer<T>() {
-                @Override
-                public void onCompleted() {
-                    if (status.get()) {
-                        observer.onNext(true);
-                        observer.onCompleted();
-                    }
-                }
+            return subscription.wrap(sequence.subscribe(new AllObserver(observer)));
 
-                @Override
-                public void onError(Exception e) {
-                    observer.onError(e);
-                }
+        }
 
-                @Override
-                public void onNext(T args) {
-                    boolean result = predicate.call(args);
-                    boolean changed = status.compareAndSet(true, result);
-
-                    if (changed && !result) {
-                        observer.onNext(false);
-                        observer.onCompleted();
-                        subscription.unsubscribe();
-                    }
+        private class AllObserver implements Observer<T> {
+            private final Observer<Boolean> underlying;
+
+            private final AtomicBoolean status = new AtomicBoolean(true);
+
+            public AllObserver(Observer<Boolean> underlying) {
+                this.underlying = underlying;
+            }
+
+            @Override
+            public void onCompleted() {
+                if (status.get()) {
+                    underlying.onNext(true);
+                    underlying.onCompleted();
                 }
-            }));
+            }
+
+            @Override
+            public void onError(Exception e) {
+                underlying.onError(e);
+            }
+
+            @Override
+            public void onNext(T args) {
+                boolean result = predicate.call(args);
+                boolean changed = status.compareAndSet(true, result);
+
+                if (changed && !result) {
+                    underlying.onNext(false);
+                    underlying.onCompleted();
+                    subscription.unsubscribe();
+                }
+            }
         }
+
     }
 
     public static class UnitTest {

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

@@ -56,37 +56,43 @@ public static <K, T> Func1<Observer<GroupedObservable<K, T>>, Subscription> grou
 
     private static class GroupBy<K, V> implements Func1<Observer<GroupedObservable<K, V>>, Subscription> {
         private final Observable<KeyValue<K, V>> source;
-        private final ConcurrentHashMap<K, Boolean> keys = new ConcurrentHashMap<K, Boolean>();
 
         private GroupBy(Observable<KeyValue<K, V>> source) {
             this.source = source;
         }
 
         @Override
         public Subscription call(final Observer<GroupedObservable<K, V>> observer) {
+            return source.subscribe(new GroupByObserver(observer));
+        }
 
-            return source.subscribe(new Observer<KeyValue<K, V>>() {
+        private class GroupByObserver implements Observer<KeyValue<K, V>> {
+            private final Observer<GroupedObservable<K, V>> underlying;
 
-                @Override
-                public void onCompleted() {
-                    observer.onCompleted();
-                }
+            private final ConcurrentHashMap<K, Boolean> keys = new ConcurrentHashMap<K, Boolean>();
 
-                @Override
-                public void onError(Exception e) {
-                    observer.onError(e);
-                }
+            private GroupByObserver(Observer<GroupedObservable<K, V>> underlying) {
+                this.underlying = underlying;
+            }
 
-                @Override
-                public void onNext(final KeyValue<K, V> args) {
-                    K key = args.key;
-                    boolean newGroup = keys.putIfAbsent(key, true) == null;
-                    if (newGroup) {
-                        observer.onNext(buildObservableFor(source, key));
-                    }
-                }
+            @Override
+            public void onCompleted() {
+                underlying.onCompleted();
+            }
+
+            @Override
+            public void onError(Exception e) {
+                underlying.onError(e);
+            }
 
-            });
+            @Override
+            public void onNext(final KeyValue<K, V> args) {
+                K key = args.key;
+                boolean newGroup = keys.putIfAbsent(key, true) == null;
+                if (newGroup) {
+                    underlying.onNext(buildObservableFor(source, key));
+                }
+            }
         }
     }
 

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

@@ -68,7 +68,6 @@ public Subscription call(Observer<T> observer) {
      * @param <T>
      */
     private static class Take<T> implements Func1<Observer<T>, Subscription> {
-        private final AtomicInteger counter = new AtomicInteger();
         private final Observable<T> items;
         private final int num;
         private final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
@@ -108,6 +107,8 @@ public void onNext(T args)
         private class ItemObserver implements Observer<T> {
             private final Observer<T> observer;
 
+            private final AtomicInteger counter = new AtomicInteger();
+
             public ItemObserver(Observer<T> observer) {
                 this.observer = observer;
             }

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

@@ -91,7 +91,6 @@ public Boolean call(T input, Integer index) {
      * @param <T>
      */
     private static class TakeWhile<T> implements Func1<Observer<T>, Subscription> {
-        private final AtomicInteger counter = new AtomicInteger();
         private final Observable<T> items;
         private final Func2<T, Integer, Boolean> predicate;
         private final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
@@ -109,6 +108,8 @@ public Subscription call(Observer<T> observer) {
         private class ItemObserver implements Observer<T> {
             private final Observer<T> observer;
 
+            private final AtomicInteger counter = new AtomicInteger();
+
             public ItemObserver(Observer<T> observer) {
                 // Using AtomicObserver because the unsubscribe, onCompleted, onError and error handling behavior
                 // needs "isFinished" logic to not send duplicated events