ensures empty inner handled properly in case of sync fusion (#3307 fo…

…llowup) Signed-off-by: Oleh Dokuka <[email protected]>
reactor · Jan 11, 2023 · 1d3a6f5 · 1d3a6f5
1 parent 36fad66
commit 1d3a6f5
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 1 deletion.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxFlattenIterable.java b/reactor-core/src/main/java/reactor/core/publisher/FluxFlattenIterable.java
@@ -588,7 +588,8 @@ void drainSync() {
 							sp = iterable.spliterator();
 							itFinite = FluxIterable.checkFinite(sp);
 
-							isEmpty = itFinite && sp.estimateSize() == 0;
+
+							isEmpty = itFinite ? sp.estimateSize() == 0 : !hasNext(sp);
 						}
 						catch (Throwable exc) {
 							resetCurrent();

diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxFlattenIterableTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxFlattenIterableTest.java
@@ -705,6 +705,24 @@ public void testFluxIterableEmptyCase(Function<Flux, Flux> fn) {
 		            .verify();
 	}
 
+	@ParameterizedTestWithName
+	@MethodSource("reactor.core.publisher.FluxIterableTest#factory")
+	public void testFluxIterableSyncFusionEmptyCase(Function<Flux, Flux> fn) {
+		Iterable<String> iterable = mock(Iterable.class);
+		Mockito.when(iterable.spliterator())
+		       .thenReturn(mock(Spliterator.class));
+
+		StepVerifier.create(
+				            Flux.just(1, 2)
+				                .flatMapIterable(__ -> iterable)
+				                .as(fn)
+				                .next()
+		            )
+		            .expectSubscription()
+		            .expectComplete()
+		            .verify();
+	}
+
 	@ParameterizedTestWithName
 	@MethodSource("reactor.core.publisher.FluxIterableTest#factory")
 	public void testFluxIterableErrorHasNext(Function<Flux, Flux> fn) {