Merge branch 'master' into multisubscribe

Conflicts:
	rxjava-core/src/main/java/rx/operators/OperatorGroupBy.java

Author: mairbek

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));
+                }
+            }
         }
     }