Fix #2181 Handle nulls from collectors in Flux.collect()

Emit an NPE early instead of signaling onNext(null).
reactor · Jun 10, 2020 · 8de5bf3 · 8de5bf3
1 parent 3daa029
commit 8de5bf3
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 0 deletions.
diff --git a/reactor-core/src/main/java/reactor/core/publisher/MonoStreamCollector.java b/reactor-core/src/main/java/reactor/core/publisher/MonoStreamCollector.java
@@ -164,6 +164,10 @@ public void onComplete() {
 				return;
 			}
 
+			if (r == null) {
+				actual.onError(Operators.onOperatorError(new NullPointerException("Collector returned null"), actual.currentContext()));
+				return;
+			}
 			complete(r);
 		}
 

diff --git a/reactor-core/src/test/java/reactor/core/publisher/MonoStreamCollectorTest.java b/reactor-core/src/test/java/reactor/core/publisher/MonoStreamCollectorTest.java
@@ -40,6 +40,7 @@
 import reactor.test.StepVerifier;
 import reactor.test.subscriber.AssertSubscriber;
 
+import static java.util.stream.Collectors.reducing;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class MonoStreamCollectorTest {
@@ -300,4 +301,14 @@ public void discardIntermediateMapOnFinisherFailure() {
 		assertThat(discardedMap).containsOnlyKeys(1, 2, 3, 4);
 	}
 
+	/**
+	 * A collector producing null should be intercepted early instead of signaling onNext(null).
+	 * @see <a href="https://github.com/reactor/reactor-core/issues/2181" target="_top">issue 2181</a>
+	 */
+	@Test
+	public void collectHandlesNulls() {
+		StepVerifier.create(Flux.empty().collect(reducing(null, (a, b) -> a)))
+				.verifyError(NullPointerException.class);
+	}
+
 }