JAVA-3060: adding new spring-reactive module for the ebook.

2021-12-15 09:22:21 +05:30
parent 1bf2293e0b
commit 0cb9125701
66 changed files with 2984 additions and 0 deletions
@@ -619,6 +619,7 @@
                <module>spring-5-reactive-security</module>
                <module>spring-5-webflux</module>
                <module>spring-5-webflux-2</module>
+                <module>spring-reactive</module>

                <module>spring-activiti</module>
                <module>spring-akka</module>
@@ -1089,6 +1090,7 @@
                <module>spring-5-reactive-oauth</module>
                <module>spring-5-reactive-security</module>
                <module>spring-5-webflux</module>
+                <module>spring-reactive</module>

                <module>spring-activiti</module>
                <module>spring-akka</module>
@@ -0,0 +1,17 @@
+## Spring Reactive
+
+This module contains articles about Spring Reactor.
+
+## Relevant articles:
+
+- [Introduction to Project Reactor Bus](https://www.baeldung.com/reactor-bus)
+- [Intro To Reactor Core](https://www.baeldung.com/reactor-core)
+- [Debugging Reactive Streams in Java](https://www.baeldung.com/spring-debugging-reactive-streams)
+- [Guide to Spring 5 WebFlux](https://www.baeldung.com/spring-webflux)
+- [Introduction to the Functional Web Framework in Spring 5](https://www.baeldung.com/spring-5-functional-web)
+- [Spring 5 WebClient](https://www.baeldung.com/spring-5-webclient)
+- [Spring WebClient vs. RestTemplate](https://www.baeldung.com/spring-webclient-resttemplate)
+- [https://www.baeldung.com/webflux-webclient-parameters](https://www.baeldung.com/webflux-webclient-parameters)
+- [Handling Errors in Spring WebFlux](https://www.baeldung.com/spring-webflux-errors)
+- [Spring Security 5 for Reactive Applications](https://www.baeldung.com/spring-security-5-reactive)
+- [https://www.baeldung.com/spring-webflux-concurrency](https://www.baeldung.com/spring-webflux-concurrency)
@@ -0,0 +1,66 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>com.baeldung</groupId>
+        <artifactId>parent-boot-2</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+        <relativePath>../parent-boot-2</relativePath>
+    </parent>
+
+    <artifactId>spring-reactive</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-security</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.reactivex.rxjava2</groupId>
+            <artifactId>rxjava</artifactId>
+            <version>${rxjava.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor.kafka</groupId>
+            <artifactId>reactor-kafka</artifactId>
+            <version>${reactor-kafka.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.security</groupId>
+            <artifactId>spring-security-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <version>${reactor.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <reactor.version>3.4.12</reactor.version>
+        <reactor-kafka.version>1.2.2.RELEASE</reactor-kafka.version>
+        <rxjava.version>2.2.19</rxjava.version>
+    </properties>
+
+</project>
@@ -0,0 +1,17 @@
+package com.baeldung.reactive.concurrency;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+* Please note we assume Mongo and Kafka are running in the local machine and on default configuration.
+* Additionally, if you want to experiment with Tomcat/Jetty instead of Netty, just uncomment the lines in pom.xml and rebuild.
+*/
+@SpringBootApplication
+public class Application {
+
+    public static void main(String[] args) {
+        SpringApplication.run(Application.class, args);
+    }
+
+}
@@ -0,0 +1,127 @@
+package com.baeldung.reactive.concurrency;
+
+import io.reactivex.Observable;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+import reactor.kafka.receiver.KafkaReceiver;
+import reactor.kafka.receiver.ReceiverOptions;
+import reactor.kafka.receiver.ReceiverRecord;
+import reactor.kafka.sender.KafkaSender;
+import reactor.kafka.sender.SenderOptions;
+import reactor.kafka.sender.SenderRecord;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@RestController
+@RequestMapping("/")
+public class Controller {
+
+    @Autowired
+    private PersonRepository personRepository;
+
+    private Scheduler scheduler = Schedulers.newBoundedElastic(5, 10, "MyThreadGroup");
+
+    private Logger logger = LoggerFactory.getLogger(Controller.class);
+
+    @GetMapping("/threads/webflux")
+    public Flux<String> getThreadsWebflux() {
+        return Flux.fromIterable(getThreads());
+    }
+
+    @GetMapping("/threads/webclient")
+    public Flux<String> getThreadsWebClient() {
+        WebClient.create("http://localhost:8080/index")
+          .get()
+          .retrieve()
+          .bodyToMono(String.class)
+          .subscribeOn(scheduler)
+          .publishOn(scheduler)
+          .doOnNext(s -> logger.info("Response: {}", s))
+          .subscribe();
+        return Flux.fromIterable(getThreads());
+    }
+
+    @GetMapping("/threads/rxjava")
+    public Observable<String> getIndexRxJava() {
+        Observable.fromIterable(Arrays.asList("Hello", "World"))
+          .map(s -> s.toUpperCase())
+          .observeOn(io.reactivex.schedulers.Schedulers.trampoline())
+          .doOnNext(s -> logger.info("String: {}", s))
+          .subscribe();
+        return Observable.fromIterable(getThreads());
+    }
+
+    @GetMapping("/threads/mongodb")
+    public Flux<String> getIndexMongo() {
+        personRepository.findAll()
+          .doOnNext(p -> logger.info("Person: {}", p))
+          .subscribe();
+        return Flux.fromIterable(getThreads());
+    }
+
+    @GetMapping("/threads/reactor-kafka")
+    public Flux<String> getIndexKafka() {
+        Map<String, Object> producerProps = new HashMap<>();
+        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
+        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        SenderOptions<Integer, String> senderOptions = SenderOptions.create(producerProps);
+        KafkaSender<Integer, String> sender = KafkaSender.create(senderOptions);
+        Flux<SenderRecord<Integer, String, Integer>> outboundFlux = Flux.range(1, 10)
+          .map(i -> SenderRecord.create(new ProducerRecord<>("reactive-test", i, "Message_" + i), i));
+        sender.send(outboundFlux)
+          .subscribe();
+
+        Map<String, Object> consumerProps = new HashMap<>();
+        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "my-consumer");
+        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
+        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
+        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        ReceiverOptions<Integer, String> receiverOptions = ReceiverOptions.create(consumerProps);
+        receiverOptions.subscription(Collections.singleton("reactive-test"));
+        KafkaReceiver<Integer, String> receiver = KafkaReceiver.create(receiverOptions);
+        Flux<ReceiverRecord<Integer, String>> inboundFlux = receiver.receive();
+        inboundFlux.subscribe(r -> {
+            logger.info("Received message: {}", r.value());
+            r.receiverOffset()
+              .acknowledge();
+        });
+        return Flux.fromIterable(getThreads());
+    }
+
+    @GetMapping("/index")
+    public Mono<String> getIndex() {
+        return Mono.just("Hello world!");
+    }
+
+    private List<String> getThreads() {
+        return Thread.getAllStackTraces()
+          .keySet()
+          .stream()
+          .map(t -> String.format("%-20s \t %s \t %d \t %s\n", t.getName(), t.getState(), t.getPriority(), t.isDaemon() ? "Daemon" : "Normal"))
+          .collect(Collectors.toList());
+    }
+}
@@ -0,0 +1,27 @@
+package com.baeldung.reactive.concurrency;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+@Document
+public class Person {
+    @Id
+    String id;
+
+    public Person(String id) {
+        this.id = id;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    @Override
+    public String toString() {
+        return "Person{" + "id='" + id + '\'' + '}';
+    }
+}
@@ -0,0 +1,6 @@
+package com.baeldung.reactive.concurrency;
+
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+
+public interface PersonRepository extends ReactiveMongoRepository<Person, String> {
+}
@@ -0,0 +1,33 @@
+package com.baeldung.reactive.debugging.consumer;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.mongo.MongoReactiveAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.security.config.web.server.ServerHttpSecurity;
+import org.springframework.security.web.server.SecurityWebFilterChain;
+import reactor.core.publisher.Hooks;
+
+import java.util.Collections;
+
+@SpringBootApplication(exclude = MongoReactiveAutoConfiguration.class)
+@EnableScheduling
+public class ConsumerDebuggingApplication {
+
+    public static void main(String[] args) {
+        Hooks.onOperatorDebug();
+        SpringApplication app = new SpringApplication(ConsumerDebuggingApplication.class);
+        app.setDefaultProperties(Collections.singletonMap("server.port", "8082"));
+        app.run(args);
+    }
+
+    @Bean
+    public SecurityWebFilterChain debuggingConsumerSpringSecurityFilterChain(ServerHttpSecurity http) {
+        http.authorizeExchange()
+            .anyExchange()
+            .permitAll();
+        http.csrf().disable();
+        return http.build();
+    }
+}
@@ -0,0 +1,152 @@
+package com.baeldung.reactive.debugging.consumer.chronjobs;
+
+import com.baeldung.reactive.debugging.consumer.model.Foo;
+import com.baeldung.reactive.debugging.consumer.model.FooDto;
+import com.baeldung.reactive.debugging.consumer.service.FooService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.MediaType;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Flux;
+
+import java.time.Duration;
+import java.util.concurrent.ThreadLocalRandom;
+
+@Component
+public class ChronJobs {
+
+    private static Logger logger = LoggerFactory.getLogger(ChronJobs.class);
+    private WebClient client = WebClient.create("http://localhost:8081");
+
+    @Autowired
+    private FooService service;
+
+    @Scheduled(fixedRate = 10000)
+    public void consumeInfiniteFlux() {
+        Flux<Foo> fluxFoo = client.get()
+            .uri("/functional-reactive/periodic-foo")
+            .accept(MediaType.TEXT_EVENT_STREAM)
+            .retrieve()
+            .bodyToFlux(FooDto.class)
+            .delayElements(Duration.ofMillis(100))
+            .map(dto -> {
+                logger.debug("process 1 with dto id {} name{}", dto.getId(), dto.getName());
+                return new Foo(dto);
+            });
+        Integer random = ThreadLocalRandom.current()
+            .nextInt(0, 3);
+        switch (random) {
+        case 0:
+            logger.info("process 1 with approach 1");
+            service.processFoo(fluxFoo);
+            break;
+        case 1:
+            logger.info("process 1 with approach 1 EH");
+            service.processUsingApproachOneWithErrorHandling(fluxFoo);
+            break;
+        default:
+            logger.info("process 1 with approach 2");
+            service.processFooInAnotherScenario(fluxFoo);
+            break;
+
+        }
+    }
+
+    @Scheduled(fixedRate = 20000)
+    public void consumeFiniteFlux2() {
+        Flux<Foo> fluxFoo = client.get()
+            .uri("/functional-reactive/periodic-foo-2")
+            .accept(MediaType.TEXT_EVENT_STREAM)
+            .retrieve()
+            .bodyToFlux(FooDto.class)
+            .delayElements(Duration.ofMillis(100))
+            .map(dto -> {
+                logger.debug("process 2 with dto id {} name{}", dto.getId(), dto.getName());
+                return new Foo(dto);
+            });
+        Integer random = ThreadLocalRandom.current()
+            .nextInt(0, 3);
+        switch (random) {
+        case 0:
+            logger.info("process 2 with approach 1");
+            service.processFoo(fluxFoo);
+            break;
+        case 1:
+            logger.info("process 2 with approach 1 EH");
+            service.processUsingApproachOneWithErrorHandling(fluxFoo);
+            break;
+        default:
+            logger.info("process 2 with approach 2");
+            service.processFooInAnotherScenario(fluxFoo);
+            break;
+
+        }
+    }
+
+    @Scheduled(fixedRate = 20000)
+    public void consumeFiniteFlux3() {
+        Flux<Foo> fluxFoo = client.get()
+            .uri("/functional-reactive/periodic-foo-2")
+            .accept(MediaType.TEXT_EVENT_STREAM)
+            .retrieve()
+            .bodyToFlux(FooDto.class)
+            .delayElements(Duration.ofMillis(100))
+            .map(dto -> {
+                logger.debug("process 3 with dto id {} name{}", dto.getId(), dto.getName());
+                return new Foo(dto);
+            });
+        logger.info("process 3 with approach 3");
+        service.processUsingApproachThree(fluxFoo);
+    }
+
+    @Scheduled(fixedRate = 20000)
+    public void consumeFiniteFluxWithCheckpoint4() {
+        Flux<Foo> fluxFoo = client.get()
+            .uri("/functional-reactive/periodic-foo-2")
+            .accept(MediaType.TEXT_EVENT_STREAM)
+            .retrieve()
+            .bodyToFlux(FooDto.class)
+            .delayElements(Duration.ofMillis(100))
+            .map(dto -> {
+                logger.debug("process 4 with dto id {} name{}", dto.getId(), dto.getName());
+                return new Foo(dto);
+            });
+        logger.info("process 4 with approach 4");
+        service.processUsingApproachFourWithCheckpoint(fluxFoo);
+    }
+
+    @Scheduled(fixedRate = 20000)
+    public void consumeFiniteFluxWitParallelScheduler() {
+        Flux<Foo> fluxFoo = client.get()
+            .uri("/functional-reactive/periodic-foo-2")
+            .accept(MediaType.TEXT_EVENT_STREAM)
+            .retrieve()
+            .bodyToFlux(FooDto.class)
+            .delayElements(Duration.ofMillis(100))
+            .map(dto -> {
+                logger.debug("process 5-parallel with dto id {} name{}", dto.getId(), dto.getName());
+                return new Foo(dto);
+            });
+        logger.info("process 5-parallel with approach 5-parallel");
+        service.processUsingApproachFivePublishingToDifferentParallelThreads(fluxFoo);
+    }
+
+    @Scheduled(fixedRate = 20000)
+    public void consumeFiniteFluxWithSingleSchedulers() {
+        Flux<Foo> fluxFoo = client.get()
+            .uri("/functional-reactive/periodic-foo-2")
+            .accept(MediaType.TEXT_EVENT_STREAM)
+            .retrieve()
+            .bodyToFlux(FooDto.class)
+            .delayElements(Duration.ofMillis(100))
+            .map(dto -> {
+                logger.debug("process 5-single with dto id {} name{}", dto.getId(), dto.getName());
+                return new Foo(dto);
+            });
+        logger.info("process 5-single with approach 5-single");
+        service.processUsingApproachFivePublishingToDifferentSingleThreads(fluxFoo);
+    }
+}
@@ -0,0 +1,22 @@
+package com.baeldung.reactive.debugging.consumer.controllers;
+
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Hooks;
+
+@RestController
+public class ReactiveConfigsToggleRestController {
+
+    @GetMapping("/debug-hook-on")
+    public String setReactiveDebugOn() {
+        Hooks.onOperatorDebug();
+        return "DEBUG HOOK ON";
+    }
+
+    @GetMapping("/debug-hook-off")
+    public String setReactiveDebugOff() {
+        Hooks.resetOnOperatorDebug();
+        return "DEBUG HOOK OFF";
+    }
+
+}
@@ -0,0 +1,27 @@
+package com.baeldung.reactive.debugging.consumer.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.util.concurrent.ThreadLocalRandom;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class Foo {
+
+    private Integer id;
+    private String formattedName;
+    private Integer quantity;
+
+    public Foo(FooDto dto) {
+        this.id = (ThreadLocalRandom.current()
+            .nextInt(0, 100) == 0) ? null : dto.getId();
+        this.formattedName = dto.getName();
+        this.quantity = ThreadLocalRandom.current()
+            .nextInt(0, 10);
+    }
+}
@@ -0,0 +1,16 @@
+package com.baeldung.reactive.debugging.consumer.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class FooDto {
+
+    private Integer id;
+    private String name;
+}
@@ -0,0 +1,44 @@
+package com.baeldung.reactive.debugging.consumer.service;
+
+import com.baeldung.reactive.debugging.consumer.model.Foo;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.ThreadLocalRandom;
+
+public class FooNameHelper {
+
+    public static Flux<Foo> concatAndSubstringFooName(Flux<Foo> flux) {
+        flux = concatFooName(flux);
+        flux = substringFooName(flux);
+        return flux;
+    }
+
+    public static Flux<Foo> concatFooName(Flux<Foo> flux) {
+        flux = flux.map(foo -> {
+            String processedName = null;
+            Integer random = ThreadLocalRandom.current()
+                .nextInt(0, 80);
+            processedName = (random != 0) ? foo.getFormattedName() : foo.getFormattedName() + "-bael";
+            foo.setFormattedName(processedName);
+            return foo;
+        });
+        return flux;
+    }
+
+    public static Flux<Foo> substringFooName(Flux<Foo> flux) {
+        return flux.map(foo -> {
+            String processedName;
+            Integer random = ThreadLocalRandom.current()
+                .nextInt(0, 100);
+
+            processedName = (random == 0) ? foo.getFormattedName()
+                .substring(10, 15)
+                : foo.getFormattedName()
+                    .substring(0, 5);
+
+            foo.setFormattedName(processedName);
+            return foo;
+        });
+    }
+
+}
@@ -0,0 +1,30 @@
+package com.baeldung.reactive.debugging.consumer.service;
+
+import com.baeldung.reactive.debugging.consumer.model.Foo;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.ThreadLocalRandom;
+
+public class FooQuantityHelper {
+
+    public static Flux<Foo> processFooReducingQuantity(Flux<Foo> flux) {
+        flux = flux.map(foo -> {
+            Integer result;
+            Integer random = ThreadLocalRandom.current()
+                .nextInt(0, 90);
+            result = (random == 0) ? result = 0 : foo.getQuantity() + 2;
+            foo.setQuantity(result);
+            return foo;
+        });
+        return divideFooQuantity(flux);
+    }
+
+    public static Flux<Foo> divideFooQuantity(Flux<Foo> flux) {
+        return flux.map(foo -> {
+            Integer result = Math.round(5 / foo.getQuantity());
+            foo.setQuantity(result);
+            return foo;
+        });
+    }
+
+}
@@ -0,0 +1,24 @@
+package com.baeldung.reactive.debugging.consumer.service;
+
+import com.baeldung.reactive.debugging.consumer.model.Foo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+
+public class FooReporter {
+
+    private static Logger logger = LoggerFactory.getLogger(FooReporter.class);
+
+    public static Flux<Foo> reportResult(Flux<Foo> input, String approach) {
+        return input.map(foo -> {
+            if (foo.getId() == null)
+                throw new IllegalArgumentException("Null id is not valid!");
+            logger.info("Reporting for approach {}: Foo with id '{}' name '{}' and quantity '{}'", approach, foo.getId(), foo.getFormattedName(), foo.getQuantity());
+            return foo;
+        });
+    }
+
+    public static Flux<Foo> reportResult(Flux<Foo> input) {
+        return reportResult(input, "default");
+    }
+}
@@ -0,0 +1,118 @@
+package com.baeldung.reactive.debugging.consumer.service;
+
+import com.baeldung.reactive.debugging.consumer.model.Foo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+import static com.baeldung.reactive.debugging.consumer.service.FooNameHelper.concatAndSubstringFooName;
+import static com.baeldung.reactive.debugging.consumer.service.FooNameHelper.substringFooName;
+import static com.baeldung.reactive.debugging.consumer.service.FooQuantityHelper.divideFooQuantity;
+import static com.baeldung.reactive.debugging.consumer.service.FooQuantityHelper.processFooReducingQuantity;
+import static com.baeldung.reactive.debugging.consumer.service.FooReporter.reportResult;
+
+@Component
+public class FooService {
+
+    private static Logger logger = LoggerFactory.getLogger(FooService.class);
+
+    public void processFoo(Flux<Foo> flux) {
+        flux = FooNameHelper.concatFooName(flux);
+        flux = FooNameHelper.substringFooName(flux);
+        flux = flux.log();
+        flux = FooReporter.reportResult(flux);
+        flux = flux.doOnError(error -> {
+            logger.error("The following error happened on processFoo method!", error);
+        });
+        flux.subscribe();
+    }
+
+    public void processFooInAnotherScenario(Flux<Foo> flux) {
+        flux = FooNameHelper.substringFooName(flux);
+        flux = FooQuantityHelper.divideFooQuantity(flux);
+        flux.subscribe();
+    }
+
+    public void processUsingApproachOneWithErrorHandling(Flux<Foo> flux) {
+        logger.info("starting approach one w error handling!");
+        flux = concatAndSubstringFooName(flux);
+        flux = concatAndSubstringFooName(flux);
+        flux = substringFooName(flux);
+        flux = processFooReducingQuantity(flux);
+        flux = processFooReducingQuantity(flux);
+        flux = processFooReducingQuantity(flux);
+        flux = reportResult(flux, "ONE w/ EH");
+        flux = flux.doOnError(error -> {
+            logger.error("Approach 1 with Error Handling failed!", error);
+        });
+        flux.subscribe();
+    }
+
+    public void processUsingApproachThree(Flux<Foo> flux) {
+        logger.info("starting approach three!");
+        flux = concatAndSubstringFooName(flux);
+        flux = reportResult(flux, "THREE");
+        flux = flux.doOnError(error -> {
+            logger.error("Approach 3 failed!", error);
+        });
+        flux.subscribe();
+    }
+
+    public void processUsingApproachFourWithCheckpoint(Flux<Foo> flux) {
+        logger.info("starting approach four!");
+        flux = concatAndSubstringFooName(flux);
+        flux = flux.checkpoint("CHECKPOINT 1");
+        flux = concatAndSubstringFooName(flux);
+        flux = divideFooQuantity(flux);
+        flux = flux.checkpoint("CHECKPOINT 2", true);
+        flux = reportResult(flux, "FOUR");
+        flux = concatAndSubstringFooName(flux).doOnError(error -> {
+            logger.error("Approach 4 failed!", error);
+        });
+        flux.subscribe();
+    }
+
+    public void processUsingApproachFourWithInitialCheckpoint(Flux<Foo> flux) {
+        logger.info("starting approach four!");
+        flux = concatAndSubstringFooName(flux);
+        flux = flux.checkpoint("CHECKPOINT 1", true);
+        flux = concatAndSubstringFooName(flux);
+        flux = divideFooQuantity(flux);
+        flux = reportResult(flux, "FOUR");
+        flux = flux.doOnError(error -> {
+            logger.error("Approach 4-2 failed!", error);
+        });
+        flux.subscribe();
+    }
+
+    public void processUsingApproachFivePublishingToDifferentParallelThreads(Flux<Foo> flux) {
+        logger.info("starting approach five-parallel!");
+        flux = concatAndSubstringFooName(flux).publishOn(Schedulers.newParallel("five-parallel-foo"))
+            .log();
+        flux = concatAndSubstringFooName(flux);
+        flux = divideFooQuantity(flux);
+        flux = reportResult(flux, "FIVE-PARALLEL").publishOn(Schedulers.newSingle("five-parallel-bar"));
+        flux = concatAndSubstringFooName(flux).doOnError(error -> {
+            logger.error("Approach 5-parallel failed!", error);
+        });
+        flux.subscribeOn(Schedulers.newParallel("five-parallel-starter"))
+            .subscribe();
+    }
+
+    public void processUsingApproachFivePublishingToDifferentSingleThreads(Flux<Foo> flux) {
+        logger.info("starting approach five-single!");
+        flux = flux.log()
+            .subscribeOn(Schedulers.newSingle("five-single-starter"));
+        flux = concatAndSubstringFooName(flux).publishOn(Schedulers.newSingle("five-single-foo"));
+        flux = concatAndSubstringFooName(flux);
+        flux = divideFooQuantity(flux);
+        flux = reportResult(flux, "FIVE-SINGLE").publishOn(Schedulers.newSingle("five-single-bar"));
+        flux = concatAndSubstringFooName(flux).doOnError(error -> {
+            logger.error("Approach 5-single failed!", error);
+        });
+        flux.subscribe();
+    }
+
+}
@@ -0,0 +1,29 @@
+package com.baeldung.reactive.debugging.server;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.security.config.web.server.ServerHttpSecurity;
+import org.springframework.security.web.server.SecurityWebFilterChain;
+import org.springframework.web.reactive.config.EnableWebFlux;
+
+import java.util.Collections;
+
+@EnableWebFlux
+@SpringBootApplication
+public class ServerDebuggingApplication {
+    
+    public static void main(String[] args) {
+        SpringApplication app = new SpringApplication(ServerDebuggingApplication.class);
+        app.setDefaultProperties(Collections.singletonMap("server.port", "8081"));
+        app.run(args);
+    }
+    
+    @Bean
+    public SecurityWebFilterChain debuggingServerSpringSecurityFilterChain(ServerHttpSecurity http) {
+        http.authorizeExchange()
+            .anyExchange()
+            .permitAll();
+        return http.build();
+    }
+}
@@ -0,0 +1,45 @@
+package com.baeldung.reactive.debugging.server.handlers;
+
+import com.baeldung.reactive.debugging.server.model.Foo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.concurrent.ThreadLocalRandom;
+
+@Component
+public class ServerHandler {
+
+    private static Logger logger = LoggerFactory.getLogger(ServerHandler.class);
+
+    public Mono<ServerResponse> useHandler(final ServerRequest request) {
+        // there are chances that something goes wrong here...
+        return ServerResponse.ok()
+            .contentType(MediaType.TEXT_EVENT_STREAM)
+            .body(Flux.interval(Duration.ofSeconds(1))
+                .map(sequence -> {
+                    logger.info("retrieving Foo. Sequence: {}", sequence);
+                    if (ThreadLocalRandom.current()
+                        .nextInt(0, 50) == 1) {
+                        throw new RuntimeException("There was an error retrieving the Foo!");
+                    }
+                    return new Foo(sequence, "name" + sequence);
+
+                }), Foo.class);
+    }
+
+    public Mono<ServerResponse> useHandlerFinite(final ServerRequest request) {
+        return ServerResponse.ok()
+            .contentType(MediaType.TEXT_EVENT_STREAM)
+            .body(Flux.range(0, 50)
+                .map(sequence -> {
+                    return new Foo(new Long(sequence), "theFooNameNumber" + sequence);
+                }), Foo.class);
+    }
+}
@@ -0,0 +1,13 @@
+package com.baeldung.reactive.debugging.server.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class Foo {
+
+    private Long id;
+    private String name;
+
+}
@@ -0,0 +1,21 @@
+package com.baeldung.reactive.debugging.server.routers;
+
+import com.baeldung.reactive.debugging.server.handlers.ServerHandler;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.function.server.RequestPredicates;
+import org.springframework.web.reactive.function.server.RouterFunction;
+import org.springframework.web.reactive.function.server.RouterFunctions;
+import org.springframework.web.reactive.function.server.ServerResponse;
+
+@Configuration
+public class ServerRouter {
+
+	@Bean
+	public RouterFunction<ServerResponse> responseRoute(@Autowired ServerHandler handler) {
+		return RouterFunctions.route(RequestPredicates.GET("/functional-reactive/periodic-foo"), handler::useHandler)
+				.andRoute(RequestPredicates.GET("/functional-reactive/periodic-foo-2"), handler::useHandlerFinite);
+	}
+
+}
@@ -0,0 +1,24 @@
+package com.baeldung.reactive.errorhandling;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.security.config.web.server.ServerHttpSecurity;
+import org.springframework.security.web.server.SecurityWebFilterChain;
+
+@SpringBootApplication
+public class ErrorHandlingApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(ErrorHandlingApplication.class, args);
+    }
+
+    @Bean
+    public SecurityWebFilterChain securityWebFilterChain(ServerHttpSecurity http) {
+        http.authorizeExchange()
+                .anyExchange()
+                .permitAll();
+        http.csrf().disable();
+        return http.build();
+    }
+}
@@ -0,0 +1,52 @@
+package com.baeldung.reactive.errorhandling;
+
+import org.springframework.boot.web.error.ErrorAttributeOptions;
+import org.springframework.boot.web.reactive.error.DefaultErrorAttributes;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+
+import java.util.Map;
+
+@Component
+public class GlobalErrorAttributes extends DefaultErrorAttributes{
+    
+    private HttpStatus status = HttpStatus.BAD_REQUEST;
+    private String message = "please provide a name";
+
+    @Override
+    public Map<String, Object> getErrorAttributes(ServerRequest request, ErrorAttributeOptions options) {
+        Map<String, Object> map = super.getErrorAttributes(request, options);
+        map.put("status", getStatus());
+        map.put("message", getMessage());
+        return map;
+    }
+
+    /**
+     * @return the status
+     */
+    public HttpStatus getStatus() {
+        return status;
+    }
+
+    /**
+     * @param status the status to set
+     */
+    public void setStatus(HttpStatus status) {
+        this.status = status;
+    }
+
+    /**
+     * @return the message
+     */
+    public String getMessage() {
+        return message;
+    }
+
+    /**
+     * @param message the message to set
+     */
+    public void setMessage(String message) {
+        this.message = message;
+    }
+}
@@ -0,0 +1,48 @@
+package com.baeldung.reactive.errorhandling;
+
+import org.springframework.boot.autoconfigure.web.WebProperties;
+import org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler;
+import org.springframework.boot.web.error.ErrorAttributeOptions;
+import org.springframework.boot.web.reactive.error.ErrorAttributes;
+import org.springframework.context.ApplicationContext;
+import org.springframework.core.annotation.Order;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerCodecConfigurer;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.BodyInserters;
+import org.springframework.web.reactive.function.server.RequestPredicates;
+import org.springframework.web.reactive.function.server.RouterFunction;
+import org.springframework.web.reactive.function.server.RouterFunctions;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Mono;
+
+import java.util.Map;
+
+@Component
+@Order(-2)
+public class GlobalErrorWebExceptionHandler extends AbstractErrorWebExceptionHandler {
+
+    public GlobalErrorWebExceptionHandler(GlobalErrorAttributes g, ApplicationContext applicationContext,
+            ServerCodecConfigurer serverCodecConfigurer) {
+        super(g, new WebProperties.Resources(), applicationContext);
+        super.setMessageWriters(serverCodecConfigurer.getWriters());
+        super.setMessageReaders(serverCodecConfigurer.getReaders());
+    }
+
+    @Override
+    protected RouterFunction<ServerResponse> getRoutingFunction(final ErrorAttributes errorAttributes) {
+        return RouterFunctions.route(RequestPredicates.all(), this::renderErrorResponse);
+    }
+
+    private Mono<ServerResponse> renderErrorResponse(final ServerRequest request) {
+
+        final Map<String, Object> errorPropertiesMap = getErrorAttributes(request, ErrorAttributeOptions.defaults());
+
+        return ServerResponse.status(HttpStatus.BAD_REQUEST)
+            .contentType(MediaType.APPLICATION_JSON)
+            .body(BodyInserters.fromValue(errorPropertiesMap));
+    }
+
+}
@@ -0,0 +1,11 @@
+package com.baeldung.reactive.errorhandling;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.web.server.ResponseStatusException;
+
+public class NameRequiredException extends ResponseStatusException {
+    
+    public NameRequiredException(HttpStatus status, String message, Throwable e) {
+        super(status, message, e);
+    }
+}
@@ -0,0 +1,28 @@
+package com.baeldung.reactive.errorhandling.handlers;
+
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Mono;
+
+@Component
+public class Handler1 {
+
+    public Mono<ServerResponse> handleRequest1(ServerRequest request) {
+        return sayHello(request).onErrorReturn("Hello, Stranger")
+            .flatMap(s -> ServerResponse.ok()
+                .contentType(MediaType.TEXT_PLAIN)
+                .bodyValue(s));
+    }
+
+    private Mono<String> sayHello(ServerRequest request) {
+        try {
+            return Mono.just("Hello, " + request.queryParam("name")
+                .get());
+        } catch (Exception e) {
+            return Mono.error(e);
+        }
+    }
+
+}
@@ -0,0 +1,37 @@
+package com.baeldung.reactive.errorhandling.handlers;
+
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Mono;
+
+@Component
+public class Handler2 {
+
+public Mono<ServerResponse> handleRequest2(ServerRequest request) {
+    return 
+        sayHello(request)
+            .flatMap(s -> ServerResponse.ok()
+                .contentType(MediaType.TEXT_PLAIN)
+                .bodyValue(s))
+            .onErrorResume(e -> sayHelloFallback()
+                .flatMap(s -> ServerResponse.ok()
+                    .contentType(MediaType.TEXT_PLAIN)
+                    .bodyValue(s)));
+}
+
+    private Mono<String> sayHello(ServerRequest request) {
+        try {
+            return Mono.just("Hello, " + request.queryParam("name")
+                .get());
+        } catch (Exception e) {
+            return Mono.error(e);
+        }
+    }
+
+    private Mono<String> sayHelloFallback() {
+        return Mono.just("Hello, Stranger");
+    }
+
+}
@@ -0,0 +1,33 @@
+package com.baeldung.reactive.errorhandling.handlers;
+
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Mono;
+
+@Component
+public class Handler3 {
+
+    public Mono<ServerResponse> handleRequest3(ServerRequest request) {
+        return 
+            sayHello(request)
+              .flatMap(s -> ServerResponse.ok()
+                        .contentType(MediaType.TEXT_PLAIN)
+                        .bodyValue(s))
+              .onErrorResume(e -> (Mono.just("Hi, I looked around for your name but found: " + 
+                    e.getMessage())).flatMap(s -> ServerResponse.ok()
+                .contentType(MediaType.TEXT_PLAIN)
+                .bodyValue(s)));
+    }
+
+    private Mono<String> sayHello(ServerRequest request) {
+        try {
+            return Mono.just("Hello, " + request.queryParam("name")
+                .get());
+        } catch (Exception e) {
+            return Mono.error(e);
+        }
+    }
+
+}
@@ -0,0 +1,29 @@
+package com.baeldung.reactive.errorhandling.handlers;
+
+import com.baeldung.reactive.errorhandling.NameRequiredException;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Mono;
+
+@Component
+public class Handler4 {
+
+public Mono<ServerResponse> handleRequest4(ServerRequest request) {
+    return ServerResponse.ok()
+        .body(sayHello(request)
+            .onErrorResume(e -> 
+                    Mono.error(new NameRequiredException(
+                        HttpStatus.BAD_REQUEST, "please provide a name", e))), String.class);
+}
+
+    private Mono<String> sayHello(ServerRequest request) {
+        try {
+            return Mono.just("Hello, " + request.queryParam("name").get());
+        } catch (Exception e) {
+            return Mono.error(e);
+        }
+    }
+
+}
@@ -0,0 +1,21 @@
+package com.baeldung.reactive.errorhandling.handlers;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Mono;
+
+@Component
+public class Handler5 {
+
+    public Mono<ServerResponse> handleRequest5(ServerRequest request) {
+        return ServerResponse.ok()
+            .body(sayHello(request), String.class);
+        
+    }
+
+    private Mono<String> sayHello(ServerRequest request) {
+            return Mono.just("Hello, " + request.queryParam("name").get());
+    }
+
+}
@@ -0,0 +1,21 @@
+package com.baeldung.reactive.errorhandling.routers;
+
+import com.baeldung.reactive.errorhandling.handlers.Handler1;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.RequestPredicates;
+import org.springframework.web.reactive.function.server.RouterFunction;
+import org.springframework.web.reactive.function.server.RouterFunctions;
+import org.springframework.web.reactive.function.server.ServerResponse;
+
+@Component
+public class Router1 {
+
+    @Bean
+    public RouterFunction<ServerResponse> routeRequest1(Handler1 handler) {
+        return RouterFunctions.route(RequestPredicates.GET("/api/endpoint1")
+            .and(RequestPredicates.accept(MediaType.TEXT_PLAIN)), handler::handleRequest1);
+    }
+
+}
--- a/Show More
+++ b/Show More