Merge #1959 into 3.3

reactor · Nov 21, 2019 · 4654b37 · 4654b37
2 parents 5b15a8e + 1435284
commit 4654b37
Show file tree

Hide file tree

Showing 2 changed files with 40 additions and 9 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java b/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java
@@ -733,26 +733,34 @@ public V poll() {
 				produced++;
 			}
 			else {
-				int p = produced;
-				if (p != 0) {
-					produced = 0;
-					GroupByMain<?, K, V> main = parent;
-					if (main != null) {
-						main.s.request(p);
-					}
-				}
+				tryReplenish();
 			}
 			return v;
 		}
 
+		void tryReplenish() {
+			int p = produced;
+			if (p != 0) {
+				produced = 0;
+				GroupByMain<?, K, V> main = parent;
+				if (main != null) {
+					main.s.request(p);
+				}
+			}
+		}
+
 		@Override
 		public int size() {
 			return queue.size();
 		}
 
 		@Override
 		public boolean isEmpty() {
-			return queue.isEmpty();
+			if (queue.isEmpty()) {
+				tryReplenish();
+				return true;
+			}
+			return false;
 		}
 
 		@Override

diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java
@@ -29,6 +29,7 @@
 import reactor.core.CoreSubscriber;
 import reactor.core.Fuseable;
 import reactor.core.Scannable;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.FluxOperatorTest;
@@ -882,4 +883,26 @@ public void scanUnicastGroupedFlux() {
 		assertThat(test.scan(Scannable.Attr.ERROR)).isSameAs(test.error);
 	}
 
+	@Test(timeout = 10000)
+	public void fusedGroupByParallel() {
+		int parallelism = 2;
+		Scheduler process = Schedulers.newParallel("process", parallelism, true);
+
+		final long start = System.nanoTime();
+
+		Flux.range(0, 500_000)
+				.subscribeOn(Schedulers.newSingle("range", true))
+				.groupBy(i -> i % 2)
+				.flatMap(g ->
+						g.key() == 0
+								? g //.hide()  /* adding hide here fixes the hang issue */
+								.parallel(parallelism)
+								.runOn(process)
+								.map(i -> i)
+								.sequential()
+								: g.map(i -> i) // no need to use hide
+				)
+				.then()
+				.block();
+	}
 }