BugFix: Another GroupBy use case found and fixed.

Author: benjchristensen

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

@@ -59,7 +59,7 @@ public void onCompleted() {
                 }
 
                 if (completionCounter.get() == 0) {
-                    // special case if no children are running (such as an empty sequence)
+                    // special case if no children are running (such as an empty sequence, or just getting the groups and not subscribing)
                     childOperator.onCompleted();
                 }
             }
@@ -88,6 +88,8 @@ public void onNext(T t) {
 
                             @Override
                             public void call(final Operator<? super T> o) {
+                                // number of children we have running
+                                completionCounter.incrementAndGet();
                                 o.add(Subscriptions.create(new Action0() {
 
                                     @Override
@@ -119,8 +121,6 @@ public void onNext(T t) {
 
                         });
                         groups.put(key, gps);
-                        // number of children we have running
-                        completionCounter.incrementAndGet();
                         childOperator.onNext(go);
                     }
                     // we have the correct group so send value to it

rxjava-core/src/test/java/rx/operators/OperatorGroupByTest.java

@@ -483,7 +483,19 @@ public Integer call(Integer i) {
                     @Override
                     public Observable<Integer> call(GroupedObservable<Integer, Integer> group) {
                         if (group.getKey() == 0) {
-                            return group.observeOn(Schedulers.newThread()).delay(200, TimeUnit.MILLISECONDS);
+                            return group.observeOn(Schedulers.newThread()).map(new Func1<Integer, Integer>() {
+
+                                @Override
+                                public Integer call(Integer t) {
+                                    try {
+                                        Thread.sleep(2);
+                                    } catch (InterruptedException e) {
+                                        e.printStackTrace();
+                                    }
+                                    return t * 10;
+                                }
+
+                            });
                         } else {
                             return group.observeOn(Schedulers.newThread());
                         }
@@ -493,6 +505,7 @@ public Observable<Integer> call(GroupedObservable<Integer, Integer> group) {
 
                     @Override
                     public void onCompleted() {
+                        System.out.println("=> onCompleted");
                         latch.countDown();
                     }
 
@@ -516,6 +529,43 @@ public void onNext(Integer s) {
         assertEquals(100, eventCounter.get());
     }
 
+    @Test
+    public void testCompletionIfInnerNotSubscribed() throws InterruptedException {
+        final CountDownLatch latch = new CountDownLatch(1);
+        final AtomicInteger eventCounter = new AtomicInteger();
+        Observable.range(0, 100)
+                .groupBy(new Func1<Integer, Integer>() {
+
+                    @Override
+                    public Integer call(Integer i) {
+                        return i % 2;
+                    }
+                })
+                .subscribe(new Observer<GroupedObservable<Integer, Integer>>() {
+
+                    @Override
+                    public void onCompleted() {
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        e.printStackTrace();
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void onNext(GroupedObservable<Integer, Integer> s) {
+                        eventCounter.incrementAndGet();
+                        System.out.println("=> " + s);
+                    }
+                });
+        if (!latch.await(500, TimeUnit.MILLISECONDS)) {
+            fail("timed out - never got completion");
+        }
+        assertEquals(2, eventCounter.get());
+    }
+
     private static class Event {
         int source;
         String message;