Error handling in Reactor-Core

RxExperiment.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.List;

public class RxExperiment {
    private static final Logger LOGGER = LoggerFactory.getLogger(RxExperiment.class);
    // https://github.com/reactor/reactor-core/issues/49#issuecomment-205367739
    public static void main(String[] args) {
        List<Integer> tickets = Arrays.asList(2, 1, 6, 8, 9, 11, 13, 15);
        Flux.fromIterable(tickets)
                .publish()  // Turn source into hot Publisher
                .autoConnect() // Instructs the hot Publisher to start when at least one `Subscriber` subscribes
                .flatMap(integer -> {
                    if (integer % 3 == 0) {
                        throw new RuntimeException("divided by 3");
                    }
                    return Mono.just("" + integer);
                }) //Business logic that might fail
                .doOnError(throwable -> {
                    LOGGER.error("Just have error {}", throwable.getMessage());
                })
                .retry() //retry on any error
                .subscribe(System.out::println);
    }
}

Output

2
1
02:01:17.569 [main] ERROR com.github.nkonev.blog.RxExperiment - Just have error divided by 3
8
02:01:17.570 [main] ERROR com.github.nkonev.blog.RxExperiment - Just have error divided by 3
11
13
02:01:17.570 [main] ERROR com.github.nkonev.blog.RxExperiment - Just have error divided by 3

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.List;

public class RxExperiment {
    private static final Logger LOGGER = LoggerFactory.getLogger(RxExperiment.class);
    // https://github.com/reactor/reactor-core/issues/49#issuecomment-205367739
    public static void main(String[] args) {
        List<Integer> tickets = Arrays.asList(2, 1, 6, 8, 9, 11, 13, 15);
        Flux.fromIterable(tickets)
                .publish()  // Turn source into hot Publisher
                .autoConnect() // Instructs the hot Publisher to start when at least one `Subscriber` subscribes
                .flatMap(integer -> Mono.fromCallable(() -> {
                    if (integer % 3 == 0) {
                        throw new RuntimeException("divided by 3");
                    }
                    return integer;
                })) //Business logic that might fail
                .doOnError(throwable -> {
                    LOGGER.error("Just have error {}", throwable.getMessage());
                })
                .retry() //retry on any error
                .subscribe(System.out::println);
    }
}

https://spring.io/blog/2018/01/16/sneak-peek-at-reactor-core-3-2-with-milestone-1

In modern io.projectreactor:reactor-core:3.3.0.RELEASE:

import reactor.core.publisher.Flux;
import java.util.Arrays;
import java.util.List;

public class RxExperiment {
    // https://github.com/reactor/reactor-core/issues/49#issuecomment-205367739
    public static void main(String[] args) {
        List<Integer> tickets = Arrays.asList(2, 1, 6, 8, 9, 11, 13, 15);
        Flux.fromIterable(tickets)
                .map(integer -> {
                    if (integer % 3 == 0) {
                        throw new RuntimeException("divided by 3");
                    }
                    return integer;
                }).onErrorContinue((throwable, o) -> process(o))
                .subscribe(System.out::println);
    }

    private static void process(Object o) {
        System.err.println("got error: " + o);
    }
}

Output:

2
1
got error: 6
8
got error: 9
11
13

In RxJava2: https://stackoverflow.com/a/28971140/4655234