Bugfix: GroupBy Completion

Found bug while doing Parallel. It was completing prematurely when child groups were asynchronous and delayed.

Author: benjchristensen

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

@@ -53,11 +53,15 @@ public Operator<? super T> call(final Operator<? super GroupedObservable<K, T>>
 
             @Override
             public void onCompleted() {
-                // if we receive onCompleted from our parent we onComplete everything
+                // if we receive onCompleted from our parent we onComplete children
                 for (PublishSubject<T> ps : groups.values()) {
                     ps.onCompleted();
                 }
-                childOperator.onCompleted();
+
+                if (completionCounter.get() == 0) {
+                    // special case if no children are running (such as an empty sequence)
+                    childOperator.onCompleted();
+                }
             }
 
             @Override
@@ -92,7 +96,25 @@ public void call() {
                                     }
 
                                 }));
-                                _gps.subscribe(o);
+                                _gps.subscribe(new Operator<T>(o) {
+
+                                    @Override
+                                    public void onCompleted() {
+                                        o.onCompleted();
+                                        completeInner();
+                                    }
+
+                                    @Override
+                                    public void onError(Throwable e) {
+                                        o.onError(e);
+                                    }
+
+                                    @Override
+                                    public void onNext(T t) {
+                                        o.onNext(t);
+                                    }
+
+                                });
                             }
 
                         });

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

@@ -466,6 +466,56 @@ public void call(String s) {
         assertEquals(37, sentEventCounter.get());
     }
 
+    @Test
+    public void testStaggeredCompletion() throws InterruptedException {
+        final AtomicInteger eventCounter = new AtomicInteger();
+        final CountDownLatch latch = new CountDownLatch(1);
+        Observable.range(0, 100)
+                .groupBy(new Func1<Integer, Integer>() {
+
+                    @Override
+                    public Integer call(Integer i) {
+                        return i % 2;
+                    }
+                })
+                .flatMap(new Func1<GroupedObservable<Integer, Integer>, Observable<Integer>>() {
+
+                    @Override
+                    public Observable<Integer> call(GroupedObservable<Integer, Integer> group) {
+                        if (group.getKey() == 0) {
+                            return group.observeOn(Schedulers.newThread()).delay(200, TimeUnit.MILLISECONDS);
+                        } else {
+                            return group.observeOn(Schedulers.newThread());
+                        }
+                    }
+                })
+                .subscribe(new Observer<Integer>() {
+
+                    @Override
+                    public void onCompleted() {
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void onError(Throwable e) {
+                        e.printStackTrace();
+                        latch.countDown();
+                    }
+
+                    @Override
+                    public void onNext(Integer s) {
+                        eventCounter.incrementAndGet();
+                        System.out.println("=> " + s);
+                    }
+                });
+
+        if (!latch.await(2000, TimeUnit.MILLISECONDS)) {
+            fail("timed out");
+        }
+
+        assertEquals(100, eventCounter.get());
+    }
+
     private static class Event {
         int source;
         String message;