1.x: fix GroupBy delaying group completion till all groups were emitted

ReactiveX · Mar 21, 2016 · c8a2cfa · c8a2cfa
1 parent 686231b
commit c8a2cfa
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 7 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorGroupBy.java b/src/main/java/rx/internal/operators/OperatorGroupBy.java
@@ -219,6 +219,12 @@ public void onCompleted() {
             if (done) {
                 return;
             }
+
+            for (GroupedUnicast<K, V> e : groups.values()) {
+                e.onComplete();
+            }
+            groups.clear();
+
             done = true;
             GROUP_COUNT.decrementAndGet(this);
             drain();
@@ -328,13 +334,6 @@ boolean checkTerminated(boolean d, boolean empty,
                     return true;
                 } else
                 if (empty) {
-                    List<GroupedUnicast<K, V>> list = new ArrayList<GroupedUnicast<K, V>>(groups.values());
-                    groups.clear();
-
-                    for (GroupedUnicast<K, V> e : list) {
-                        e.onComplete();
-                    }
-
                     actual.onCompleted();
                     return true;
                 }

diff --git a/src/test/java/rx/GroupByTests.java b/src/test/java/rx/GroupByTests.java
@@ -21,6 +21,7 @@
 import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.observables.GroupedObservable;
+import rx.observers.TestSubscriber;
 
 public class GroupByTests {
 
@@ -90,4 +91,27 @@ public void call(String v) {
 
         System.out.println("**** finished");
     }
+
+    @Test
+    public void groupsCompleteAsSoonAsMainCompletes() {
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+
+        Observable.range(0, 20)
+        .groupBy(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer i) {
+                return i % 5;
+            }
+        })
+        .concatMap(new Func1<GroupedObservable<Integer, Integer>, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(GroupedObservable<Integer, Integer> v) {
+                return v;
+            }
+        }).subscribe(ts);
+
+        ts.assertValues(0, 5, 10, 15, 1, 6, 11, 16, 2, 7, 12, 17, 3, 8, 13, 18, 4, 9, 14, 19);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
 }