Merge pull request #1966 from zsxwing/fix-groupby-null-key

Fix NPE when the key is null in GroupBy

Author: benjchristensen

src/main/java/rx/internal/operators/OperatorGroupBy.java

@@ -102,7 +102,7 @@ public Observer<T> getObserver() {
 
         }
 
-        private final ConcurrentHashMap<K, GroupState<K, T>> groups = new ConcurrentHashMap<K, GroupState<K, T>>();
+        private final ConcurrentHashMap<Object, GroupState<K, T>> groups = new ConcurrentHashMap<Object, GroupState<K, T>>();
 
         private static final NotificationLite<Object> nl = NotificationLite.instance();
 
@@ -166,10 +166,18 @@ void requestFromGroupedObservable(long n, GroupState<K, T> group) {
             }
         }
 
+        private Object groupedKey(K key) {
+            return key == null ? NULL_KEY : key;
+        }
+
+        private K getKey(Object groupedKey) {
+            return groupedKey == NULL_KEY ? null : (K) groupedKey;
+        }
+
         @Override
         public void onNext(T t) {
             try {
-                final K key = keySelector.call(t);
+                final Object key = groupedKey(keySelector.call(t));
                 GroupState<K, T> group = groups.get(key);
                 if (group == null) {
                     // this group doesn't exist
@@ -185,10 +193,10 @@ public void onNext(T t) {
             }
         }
 
-        private GroupState<K, T> createNewGroup(final K key) {
+        private GroupState<K, T> createNewGroup(final Object key) {
             final GroupState<K, T> groupState = new GroupState<K, T>();
 
-            GroupedObservable<K, R> go = GroupedObservable.create(key, new OnSubscribe<R>() {
+            GroupedObservable<K, R> go = GroupedObservable.create(getKey(key), new OnSubscribe<R>() {
 
                 @Override
                 public void call(final Subscriber<? super R> o) {
@@ -252,7 +260,7 @@ public void onNext(T t) {
             return groupState;
         }
 
-        private void cleanupGroup(K key) {
+        private void cleanupGroup(Object key) {
             GroupState<K, T> removed;
             removed = groups.remove(key);
             if (removed != null) {
@@ -357,4 +365,5 @@ public Object call(Object t) {
         }
     };
 
+    private static final Object NULL_KEY = new Object();
 }

src/test/java/rx/internal/operators/OperatorGroupByTest.java

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -1357,4 +1358,31 @@ public Observable<Integer> call(GroupedObservable<Integer, Integer> t) {
 
     };
 
-}
+    @Test
+    public void testGroupByWithNullKey() {
+        final String[] key = new String[]{"uninitialized"};
+        final List<String> values = new ArrayList<String>();
+        Observable.just("a", "b", "c").groupBy(new Func1<String, String>() {
+
+            @Override
+            public String call(String value) {
+                return null;
+            }
+        }).subscribe(new Action1<GroupedObservable<String, String>>() {
+
+            @Override
+            public void call(GroupedObservable<String, String> groupedObservable) {
+                key[0] = groupedObservable.getKey();
+                groupedObservable.subscribe(new Action1<String>() {
+
+                    @Override
+                    public void call(String s) {
+                        values.add(s);
+                    }
+                });
+            }
+        });
+        assertEquals(null, key[0]);
+        assertEquals(Arrays.asList("a", "b", "c"), values);
+    }
+}