ensures empty inner handled properly in case of sync fusion (#3329)

closes #3327 (#3307 followup)
reactor · Mar 7, 2023 · 76c3f0a · 76c3f0a
1 parent 6e982f1
commit 76c3f0a
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 3 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxFlattenIterable.java b/reactor-core/src/main/java/reactor/core/publisher/FluxFlattenIterable.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -588,7 +588,7 @@ 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
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -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) {