1.x: fix groupBy consuming the upstream in an unbounded manner (ReactiveX#5030)

Author: akarnokd

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

@@ -202,7 +202,7 @@ public void onNext(T t) {
                 return;
             }
 
-            boolean notNew = true;
+            boolean newGroup = false;
             Object mapKey = key != null ? key : NULL_KEY;
             GroupedUnicast<K, V> group = groups.get(mapKey);
             if (group == null) {
@@ -214,9 +214,7 @@ public void onNext(T t) {
 
                     groupCount.getAndIncrement();
 
-                    notNew = false;
-                    q.offer(group);
-                    drain();
+                    newGroup = true;
                 } else {
                     return;
                 }
@@ -243,8 +241,9 @@ public void onNext(T t) {
                 }
             }
 
-            if (notNew) {
-                s.request(1);
+            if (newGroup) {
+                q.offer(group);
+                drain();
             }
         }
 

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

@@ -2017,4 +2017,34 @@ public Map<Integer, Object> call(Action1<Integer> t) {
                 throw exception;
             }};
     }
+    
+    @Test
+    public void outerConsumedInABoundedManner() {
+        final int[] counter = { 0 };
+        
+        Observable.range(1, 10000)
+        .doOnRequest(new Action1<Long>() {
+            @Override
+            public void call(Long v) {
+                counter[0] += v;
+            }
+        })
+        .groupBy(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return 1;
+            }
+        })
+        .flatMap(new Func1<GroupedObservable<Integer, Integer>, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(GroupedObservable<Integer, Integer> v) {
+                return v;
+            }
+        })
+        .test(0);
+
+        int c = counter[0];
+        assertTrue("" + c, c > 0);
+        assertTrue("" + c, c < 10000);
+    }
 }