From 93637134e9ac0cc8d84eafd02465779b444d6324 Mon Sep 17 00:00:00 2001
From: Nguyen Nam Thai <namthainguyen@gmail.com>
Date: Sun, 1 Aug 2021 23:56:06 +0700
Subject: [PATCH] BAEL-4883 Exception handling in Project Reactor

---
 .../reactor/exception/ExceptionUnitTest.java  | 89 +++++++++++++++++++
 1 file changed, 89 insertions(+)
 create mode 100644 reactor-core/src/test/java/com/baeldung/reactor/exception/ExceptionUnitTest.java

diff --git a/reactor-core/src/test/java/com/baeldung/reactor/exception/ExceptionUnitTest.java b/reactor-core/src/test/java/com/baeldung/reactor/exception/ExceptionUnitTest.java
new file mode 100644
index 0000000000..f4da2e325a
--- /dev/null
+++ b/reactor-core/src/test/java/com/baeldung/reactor/exception/ExceptionUnitTest.java
@@ -0,0 +1,89 @@
+package com.baeldung.reactor.exception;
+
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SynchronousSink;
+import reactor.test.StepVerifier;
+
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+public class ExceptionUnitTest {
+    @Test
+    public void givenInputStreamWithAnInvalidElement_whenAPipelineOperatorThrowsAnException_thenAnErrorIsSentDownstream() {
+        Function<String, Integer> mapper = input -> {
+            if (input.matches("\\D")) {
+                throw new NumberFormatException();
+            } else {
+                return Integer.parseInt(input);
+            }
+        };
+
+        Flux<String> inFlux = Flux.just("1", "1.5", "2");
+        Flux<Integer> outFlux = inFlux.map(mapper);
+
+        StepVerifier.create(outFlux)
+                .expectNext(1)
+                .expectError(NumberFormatException.class)
+                .verify();
+    }
+
+    @Test
+    public void givenInputStreamWithAnInvalidElement_whenTheHandleOperatorCallsSinkErrorMethod_thenAnErrorIsSentDownstream() {
+        BiConsumer<String, SynchronousSink<Integer>> handler = (input, sink) -> {
+            if (input.matches("\\D")) {
+                sink.error(new NumberFormatException());
+            } else {
+                sink.next(Integer.parseInt(input));
+            }
+        };
+
+        Flux<String> inFlux = Flux.just("1", "1.5", "2");
+        Flux<Integer> outFlux = inFlux.handle(handler);
+
+        StepVerifier.create(outFlux)
+                .expectNext(1)
+                .expectError(NumberFormatException.class)
+                .verify();
+    }
+
+    @Test
+    public void givenInputStreamWithAnInvalidElement_whenTheFlatMapOperatorCallsMonoErrorMethod_thenAnErrorIsSentDownstream() {
+        Function<String, Publisher<Integer>> mapper = input -> {
+            if (input.matches("\\D")) {
+                return Mono.error(new NumberFormatException());
+            } else {
+                return Mono.just(Integer.parseInt(input));
+            }
+        };
+
+        Flux<String> inFlux = Flux.just("1", "1.5", "2");
+        Flux<Integer> outFlux = inFlux.flatMap(mapper);
+
+        StepVerifier.create(outFlux)
+                .expectNext(1)
+                .expectError(NumberFormatException.class)
+                .verify();
+    }
+
+    @Test
+    public void givenInputStreamWithANullElement_whenAPipelineOperatorIsCalled_thenAnNpeIsSentDownstream() {
+        Function<String, Integer> mapper = input -> {
+            if (input == null) {
+                return 0;
+            } else {
+                return Integer.parseInt(input);
+            }
+        };
+
+        Flux<String> inFlux = Flux.just("1", null, "2");
+        Flux<Integer> outFlux = inFlux.map(mapper);
+
+        StepVerifier.create(outFlux)
+                .expectNext(1)
+                .expectError(NullPointerException.class)
+                .verify();
+    }
+}