From c8a2cfa78c025bed0d6c499f80ffb2d13dc918f4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?D=C3=A1vid=20Karnok?= <akarnokd@gmail.com>
Date: Mon, 21 Mar 2016 20:10:18 +0100
Subject: [PATCH] 1.x: fix GroupBy delaying group completion till all groups
 were emitted

---
 .../internal/operators/OperatorGroupBy.java   | 13 +++++-----
 src/test/java/rx/GroupByTests.java            | 24 +++++++++++++++++++
 2 files changed, 30 insertions(+), 7 deletions(-)

diff --git a/src/main/java/rx/internal/operators/OperatorGroupBy.java b/src/main/java/rx/internal/operators/OperatorGroupBy.java
index 38edc0a68f..4fe29b6c2d 100644
--- 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
index 3530c08799..a4527777ef 100644
--- 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();
+    }
 }